Spring整合JMS(消息中间件)

Posted 技术专家

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Spring整合JMS(消息中间件)相关的知识，希望对你有一定的参考价值。

这一节来说说，异步机制及spring对JMS封装

 

 

 

一、消息异步处理

类似于RMI、Hessian、Burlap等远程方法调用，它们都是同步的，所谓同步调用就是客户端必须等待操作完成，如果远程服务没有返回任何响应，客户端会一直等待直到服务完成。

所谓同步：就是客户端必须等待操作完成，如果远程服务没有任何返回响应，客户端会一直等待直到服务完成。

异步调用：客户端发送消息无需等待服务处理完成便可立即返回，就像发送完消息就立刻被处理成功一样。

 

1.1消息的 发送

异步处理的世界，我们可以把消息发送比作一个邮局系统。我们不必关心邮件如何送出，能否到达，邮局系统会保证邮件最终送达我们希望的接受者手中，点对点   和发布、订阅。

 

1.1点对点模式

在点对点模式中，每个消息只有一个发送者和一个接受者，

在点对点模型中，每个消息只有一个发送者和一个接收者。如下图所示：
技术分享

 

 

在点对点模型中， 消息broker会把消息放入一个queue。当一个接收者请求下一个消息时，消息会被从queue中取出并传递给接收者。因为消息从queue中取出便会被移除，所以这保证了一个消息只能有一个接收者。

尽管消息队列中的每个消息只有一个接收者，但这并不意味着只能有一个接收者从队列获取消息，可以同时有多个接收者从队列获取消息，只不过它们只能处理各自接收到的消息。其实这就像在银行排队一样，排队的人可以看做一个个消息，而银行工作窗口便是消息的接收者，每个窗口服务完一个客户之后都会让队列中的“下一个”到窗口办理业务。

还有，如果多个接收者监听一个队列，我们是很难确定到底哪个接收者处理哪个消息的。不过这也不一定不好，因为这样就使得我们很方便的通过增加接收者来拓展应用处理能力了。

 

 

1.2发布/订阅模式

 

在发布/订阅模式中，消息是被发送到topic中的。就像queue一样，很多接收者可以监听同一个topic，但是与queue每个消息只传递给一个接收者不同，订阅了同一个topic的所有接收者都会收到消息的拷贝，如下图所示：
技术分享
从发布/订阅的名字中我们也可看出，发布者发布一条消息，所有订阅者都能收到，这就是发布订阅模式最大的特性。对于发布者来说，它只知道将消息发布到了一个特定的topic，它不关心谁监听这个topic，这也就意味着它并不知道这些消息是被如何处理的。

 

1.2 异步消息系统带来的好处

在具体介绍异步消息系统带来的好处之前，我们先看看同步系统的局限性：

同步会话意味着等待：当客户的调用远程服务的方法时，客户端必须等待远程方法结束之后才能继续，如果客户端与远程服务交流频繁或者远程服务响应过慢，会影响客户端的性能 客户端和服务接口耦合：如果服务接口发生改变，所有客户的都需要修改 客户端和服务位置耦合：客户端要想使用远程服务就必须配置服务的地址，如果网络拓扑发生变化，客户端需要重新配置服务地址 客户端和服务可用性耦合：如果服务不可用，那么也会导致客户端不可用

下面我们再看一下异步消息系统是如何解决这些问题的。

无需等待

当一个消息被异步发送，客户端不需要等待它处理完成。客户端直接把消息扔给broker然后做其它事情，broker负责把消息送到合适的目的地。

因为客户端不需要等待，所以客户端的性能会有很大的提升。

面向消息和解耦合

不同于传统基于方法调用的RPC会话，消息异步发送是以数据为中心的。这就意味着客户端不需要和某个方法签名绑定，任何queue或topic的订阅者都可以处理客户端发送的消息。客户端不必再关心服务方任何相关的问题。

位置独立

同步RPC服务的调用是通过网络地址定位的，这就意味着客户端无法摆脱网络拓扑的变化。如果服务的IP或端口发生改变，客户端也需要做相应的改变。

相反，异步消息系统中的客户端并不关心服务所在的位置及其如何处理消息，它只负责将消息发送到特定的queue或topic。所以，服务位于什么地方都无所谓，只要它们能够从queue或topic中获取消息即可。

在点对点模式中，可以很方便的利用位置独立这个特性创建一个服务集群。客户端不需要关心服务的位置，集群中各个服务仅需知道broker的位置，并从同一个queue获取消息，如果服务压力过大无法及时处理消息，我们只需要在集群中增加一个服务实例去监听同一个queue即可。

在发布/订阅模式中，位置独立同样有很重要的作用。多个服务可以订阅同一个topic，他们都能获取到topic中的每个消息，但是对各个服务的处理可以不同。比如我们有一个服务集合订阅了一个接收新员工消息的topic，所以这些服务都可以得到每个新员工消息，一个服务可以将新员工添加到薪资系统，另一个服务可以将新员工增加到hr系统，还有服务负责赋予新员工各种系统权限等等，每个订阅topic的服务都能对各自的消息做出自己的处理。

可靠性保证

当一个客户端和服务通过同步方式进行交互时，如果服务出现任何问题挂掉，都会影响客户端正常工作。但是当消息是异步发送时，客户端与服务之间被broker隔离，客户端只负责发送消息，即使当发送消息时服务挂掉，消息也会被broker存储起来，等到服务可用时再接着进行处理。

二、通过JMS发送消息

Java Message Service是一个Java标准，它定义了一套与消息broker交互的通用API。在JMS出现之前，每一种消息broker都有自己独特的一套API，使得应用代码无法在不同的broker之间适用。但是通过JMS，所有与broker交互的代码就可以适用一套通用的API，就像JDBC一样。

当然Spring对JMS也提供了支持，即JmsTemplate。通过JmsTemplate，我们可以更加方便地向queue和topic发送和接收消息。后面我们会详细介绍Spring对JMS的实现，但是在发送和接收消息之前，我们需要现有一个broker。

2.1 在Spring中配置消息broker

ActiveMQ是非常优秀的JMS框架，关于ActiveMQ相关内容这里不多做介绍，具体可以参考：http://activemq.apache.org/，本篇主要介绍如何在Spring中对其进行配置和使用。

2.1.1 创建一个connection factory

我们要想发送消息到ActiveMQ，就需要先创建到它的连接，ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类。在Spring中配置方式如下：

?
1
2
<code><bean class="org.apache.activemq.spring.ActiveMQConnectionFactory" id="connectionFactory" p:brokerurl="tcp://localhost:61616">
</bean></code>

除此之外，Spring为ActiveMQ提供了专门的命名空间，我们可以使用Spring的ActiveMQ命名空间来创建连接工厂。首先要在配置文件中声明amq命名空间：

?
1
2
3
4
5
6
7
8
9
10
<code><!--?xml version="1.0" encoding="UTF-8"?-->
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:jms="http://www.springframework.org/schema/jms" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemalocation="http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">
...
</beans>
</code>

然后我们就可以利用元素来声明一个连接工厂：

?
1
2
<code>
</amq:connectionfactory></code>

需要注意，元素是专门针对ActiveMQ的。如果我们用到的是其它broker，就需要用另外的标签元素或注入另外的工厂bean。上面元素中的brokerURL指定了ActiveMQ在服务器中的IP和端口，上面端口值就是ActiveMQ默认端口。

 

2.1.2 声明ActiveMQ的消息destination

除了要有一个连接工厂之外，我们还需要知道消息发送到的destination。上面讲过了，消息的destination只有两类queue或者topic，在Spring中，我们需要配置queue或topic对应的bean。

配置一个ActiveMQ queue bean：

?
1
2
<code><bean c:_="biz1.queue" class="org.apache.activemq.command.ActiveMQQueue" id="queue">
</bean></code>

配置一个ActiveMQ topic bean：

?
1
2
<code><bean c:_="biz1.topic" class="org.apache.activemq.command.ActiveMQTopic" id="topic">
</bean></code>

上面例子中c:_属性代表的是构造器参数，它指定了queue或topic的名称。

像连接工厂一样，Spring提供了另外一种配置destination的方式，就是通过Spring ActiveMQ命名空间进行配置。

使用元素配置一个queue：

?
1
2
<code>
</amq:queue></code>

使用元素配置一个topic:

?
1
2
<code>
</amq:topic></code>

上面元素中physicalName属性代表消息通道的名称，也就是queue和topic的名称。

通过上面两个组件的配置，我们就可以向ActiveMQ发送和接收消息了。发送和接收消息我们使用的是Spring提供的JmsTempate，它是Spring对JMS的抽象，下面就详细介绍JMSTemplate的使用。

 

2.2 使用Spring的JMS template

虽然JMS提供了一套与各种broker交互的通用API，但实际使用起来并不是很方便，我们先看一下使用普通JMS API与broker交互的代码。

2.2.1 通过普通JMS API发送消息到broker代码：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
<code>ConnectionFactory cf =new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection conn = null;
Session session = null;
try {
    conn = cf.createConnection();
    session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Destination destination = new ActiveMQQueue("spitter.queue");
    MessageProducer producer = session.createProducer(destination);
    TextMessage message = session.createTextMessage();
    message.setText("Hello world!");
    producer.send(message);
} catch (JMSException e) {
    // handle exception?
} finally {
    try {
        if (session != null) {
            session.close();
        }
        if (conn != null) {
            conn.close();
        }
    } catch (JMSException ex) {
    }
}
</code>

上面代码中我们可以看到，为了发送一条 “Hello world”的消息却用了20多行代码，就像JDBC一样，我们大部分代码都是再做一些重复性的准备工作，比如获取连接、创建session、异常处理等等。其实接收消息的代码也是如此，在JDBC中，Spring提供了一个JdbcTemplate来简化JDBC代码开发，同样，Spring也提供了JmsTemplate来简化JMS消息处理的开发。

 

2.2.2 使用JmsTemplate

JmsTemplate其实是Spring对JMS更高一层的抽象，它封装了大部分创建连接、获取session及发送接收消息相关的代码，使得我们可以把精力集中在消息的发送和接收上。另外，JmsTemplate对异常也做了很好的封装，其对应的基本的异常为JMSException。

要使用JmsTemplate，就要在Spring配置文件中配置它作为一个bean：

?
1
2
<code><bean c:_-ref="connectionFactory" class="org.springframework.jms.core.JmsTemplate" id="jmsTemplate">
</bean></code>

因为JmsTemplate需要先和broker进行连接，所以它需要依赖一个connectionFactory。

 

因为JmsTemplate需要先和broker进行连接，所以它需要依赖一个connectionFactory。

发送消息

假如我们有一个业务需要用到异步消息发送，我们先定义这样一个业务接口：

?
1
2
3
4
<code>public interface MyMessageService {
    void sendMessage(String message);
}
</code>

上面接口中只有一个方法，就是发送消息。

我们写这个接口的实现，在这个接口实现中，我们就是用JmsTemplate实现异步消息发送：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
<code>import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsOperations;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
 
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
 
/**
 * Created by [email protected] on 2016/6/17.
 */
@Component
public class MyMessageServiceImpl implements MyMessageService{
    @Autowired
    private JmsOperations jmsOperations;
    public void sendMessage(final String message) {
        jmsOperations.send("biz1.queue", new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(message);
            }
        });
    }
}
</code>

我们可以看到，我们业务的实现中注入了一个JmsOperations对象，这个对象就是JmsTempate的实现。JmsOperations的send()方法有两个参数，第一个是消息的destination，第二个便是具体的Message，在上面例子中message是通过一个匿名内部类MessageCreator的createMessage()方法构造的。

通过上面例子可以发现，通过JmsTempate，我们只需要关心发送消息即可，所有的连接和session的维护都由JmsTempate负责。

设置默认destination

大部分情况下，一个业务消息的destination是相同的，所以我们不必每次发送都填写destination，我们可以在配置文件中对其进行配置：

?
1
2
<code><bean c:_-ref="connectionFactory" class="org.springframework.jms.core.JmsTemplate" id="jmsTemplate" p:defaultdestinationname="biz1.queue">
</bean></code>

在上面配置中我们默认destination值为biz1.queue，因为它只是声明了一个名称，并没有说明是哪种类型的destination，所以，如果存在相同名称的queue或topic，就会自动与之匹配，如果不存在，则会默认创建一个相同名称的queue。如果我们想指定destination的类型，我们可以通过配置让其依赖之前配置的destination bean即可：

?
1
2
<code><bean c:_-ref="connectionFactory" class="org.springframework.jms.core.JmsTemplate" id="jmsTemplate" p:defaultdestination-ref="biz1.Topic">
</bean></code>

当我们配置了默认destination，我们就可以在发送消息时省略第一个参数了：

?
1
2
3
4
5
6
<code>jmsOperations.send(
    new MessageCreator() {
        ...
    }
);
</code>

其实上面的send()方法可以变得更简单，我们可以利用消息转换器。

使用消息转换器发送消息

除了send()方法之外，JmsTemplate还提供了convertAndSend()方法。与send()方法需要依赖一个MessageCreator不同，convertAndSend()方法只需要传入你想发送的消息即可。下面我们用convertAndSend()实现接口中的sendMessage()方法：

?
1
2
3
4
<code>public void sendMessage(final String message) {
    jmsOperations.convertAndSend(message);
}
</code>

convertAndSend()方法会自动把你发的消息转换成Message，具体如何转换的由org.springframework.messaging.converter.MessageConverter的实现来决定。我们先看一下MessageConverter接口：

?
1
2
3
4
5
<code>public interface MessageConverter {
    Object fromMessage(Message<!--?--> var1, Class<!--?--> var2);
    Message<!--?--> toMessage(Object var1, MessageHeaders var2);
}
</code>

我们可以看到这个接口中只有两个方法而且很容易实现。其实大部分情况下我们不需要自己去实现这个接口，Spring已经为我们提供给了很多常用的实现：

默认情况下，当JmsTemplate的convertAndSend()方法使用的是SimpleMessageConverter。但是我们也可以通过配置把我们自定义的MessageConverter作为属性注入到JmsTemplate中，比如我们有个一MessageConverter的实现bean：

?
1
2
<code><bean class="org.springframework.jms.support.converter.MappingJacksonMessageConverter" id="messageConverter">
</bean></code>

我们可以把上面这个bean注入到JmsTemplate中：

?
1
2
<code><bean c:_-ref="connectionFactory" class="org.springframework.jms.core.JmsTemplate" id="jmsTemplate" p:defaultdestinationname="spittle.alert.queue" p:messageconverter-ref="messageConverter">
</bean></code>

消费消息

对于消费来说，JmsTemplate使用起来比发送更简单，只需要调用JmsOperations的receive()方法即可：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
<code>public class ReceiveMessage {
    @Autowired
    private JmsOperations jmsOperations;
    public String receive() {
        try {
            ObjectMessage message = (ObjectMessage) jmsOperations.receive();
            return (String) message.getObject();
        } catch (JMSException e) {
            e.printStackTrace();
            throw JmsUtils.convertJmsAccessException(e);
        }
    }
}
</code>

当调用 jmsOperations.receive()方法时，它会尝试从broker获取消息，若此时没有消息，receive()方法会一直等待直到有消息产生。前面例子中，当我们发送消息的时候消息被封装成的是ObjectMessage，所我们在获取的时候可以再将其转换回ObjectMessage。

这里有一点需要注意，当调用message.getObject()方法时会抛出JMSException，这个异常是属于JMS API的。JMSException是一个检查异常，在JMS操作中会抛出各种各样的JMSException，但是前面我们使用JmsTemplate时并没有捕获任何JMSException，是因为JmsTemplate内部已经将需要检查的JMSException转换成了非检查的Spring自己的JmsException。在上面代码中因为调用的是message.getObject()方法而不是JmsTemplate的方法，所以我们需要捕获JMSException。但是按照Spring的设计理念，我们应该尽量减少检查异常，所以在catch块里面我们又通过JmsUtils工具把JMSException转换成了非检查的JmsException。

同样，就行消息的发送一样，我们也可以使用JmsTemplate的receiveAndConvert()方法替换receive()方法：

?
1
2
3
4
<code>public String receive() {
    return (String)jmsOperations.receiveAndConvert();
}
</code>

我们看到，因为使用的是JmsTemplate的方法，所以我们不需要再捕获JMSException检查异常。
技术分享
不管使用msTemplate的receive()还是receiveAndConvert()方法消费消息，它们都是同步的。也就是说接收者在消息到达时需要等待。这样看起来是不是有点奇怪？发送消息时是异步的，接收消息时却是同步的。<喎?"/kf/ware/vc/" target="_blank" class="keylink">vcD4NCjxwPtXi0rK+zcrHzqrKssO0u+HT0M/Cw+a1xM/7z6LH/bavUE9KT7P2z9a1xNSt0vKjrM/Cw+bO0sPHvs2/tNK7z8LI57rOyrXP1tLssr21xL3TytXP+8+ioaM8L3A+DQo8aDMgaWQ9"23-创建消息驱动pojo">2.3 创建消息驱动POJO

我们上面已经知道，JmsTemplate的receive()方法是一个同步方法，在消息到达之前这个方法会挂起一直等待直到消息出现，如果这样的话，我们的应用可能会出现一直等待消息而不能做其它事情的情况。为何不让应用先去处理其它业务，当消息出现时再告知应用处理呢？

在EJB中，message driven bean（MDB）就可以实现异步的处理消息。Spring在这方面参考了EJB3对MDB的实现，不过在Spring中我们把它称作消息驱动POJO，也就是message-driven POJO（MDP）。

 

2.3.1 创建一个消息监听器

要想在消息出现时得到通知，那么就需要一个监听器监听queue或者topic，之所以称作消息驱动POJO，意识因为监听器是消息驱动的，而是因为这个监听器本身就是一个普通的POJO对象，不需要依赖任何接口：

?
1
2
3
4
5
6
<code>public class MyMessageHandler {
    public void handleMessage(String message){
        //具体的实现
    }
}
</code>

有了这个POJO对象，下面只需要做简单的配置即可。

2.3.2 配置消息监听器

赋予上面POJO接收消息能力的关键在于将其配置成一个Spring消息监听器，Spring的jms命名空间提供了所有相关配置。

首先，我们现需要把上面的POJO对象声明成一个bean：

?
1
2
<code><bean class="com.heaven.springexamples.jms.MyMessageHandler" id="myMessageHandler">
</bean></code>

其次，把MessageHandler变成一个消息驱动POJO，即把这个bean声明成一个listener：

?
1
2
3
4
<code><jms:listener-container connection-factory="connectionFactory">
<jms:listener destination="biz1.queue" method="handleMessage" ref="myMessageHandler">
</jms:listener></jms:listener-container>
</code>

通过上面配置，消息监听容器里面就多了一个消息监听器。消息监听容器是一个特殊的bean，它能够监听JMS的destination，监听消息的到达。一旦消息到达，消息监听容器会接受这个消息并将其发送给所有相关的listener。下面这幅图展示了整个内部处理过程：
技术分享
为了配置监听容器和监听者，我们用到了jms命名空间中的两个元素。是父元素，是子元素。依赖一个connectionFactory，这样它的各个就可以监听消息了。用来定义具体接收消息的bean及方法。按照上面的配置，当消息到达queue时，MyMessageHandler的handleMessage方法便会被调用。

2.3.3 另一种方式，实现一个MessageListener接口

需要注意到是，我们的MessageHandler还可以实现一个MessageListener接口，这样的话就不需要再单独指定消息处理的方法了，MyMessageHandler的onMessage()方法会自动被调用。MessageListener接口定义如下：

?
1
2
3
4
<code>public interface MessageListener {
    void onMessage(Message var1);
}
</code>

我们写一个简单的实现类：

?
1
2
3
4
5
6
<code>public class MyMessageListener implements MessageListener{
    public void onMessage(Message message) {
        //具体的实现
    }
}
</code>

然后直接配置listener即可（不用再配置method方法属性）：
 

?
1
2
3
<jms:listener-container connection-factory="connectionFactory">
<jms:listener destination="biz1.queue" ref="myMessageHandler">
</jms:listener></jms:listener-container>

 









以上是关于Spring整合JMS(消息中间件)的主要内容，如果未能解决你的问题，请参考以下文章

Spring整合JMS(消息中间件)

消息中间件ActiveMQ及Spring整合JMS的介绍

深入浅出JMS--Spring和ActiveMQ整合的完整实例

深入浅出JMS--Spring和ActiveMQ整合的完整实例

spring整合activemq

Java消息队列-Spring整合ActiveMq