Spring整合JMS(一)——基于ActiveMQ实现

分手后的思念是犯贱 2022-02-26 13:58 166阅读 0赞

# Spring整合JMS(一)——基于ActiveMQ实现 #

## ***1.1* *JMS*简介** ##

*JMS*的全称是*Java Message Service*，即*Java*消息服务。它主要用于在生产者和消费者之间进行消息传递，生产者负责产生消息，而消费者负责接收消息。把它应用到实际的业务需求中的话我们可以在特定的时候利用生产者生成一消息，并进行发送，对应的消费者在接收到对应的消息后去完成对应的业务逻辑。对于消息的传递有两种类型，一种是点对点的，即一个生产者和一个消费者一一对应；另一种是发布*/*订阅模式，即一个生产者产生消息并进行发送后，可以由多个消费者进行接收。

## ***1.2* *Spring*整合*JMS*** ##

对*JMS*做了一个简要介绍之后，接下来就讲一下*Spring*整合*JMS*的具体过程。*JMS*只是一个标准，真正在使用它的时候我们需要有它的具体实现，这里我们就使用*Apache*的*activeMQ*来作为它的实现。所使用的依赖利用*Maven*来进行管理，具体依赖如下：

<dependencies>
 <dependency>
 <groupId>junit</groupId>
 <artifactId>junit</artifactId>
 <version>4.10</version>
 <scope>test</scope>
 </dependency>
 <dependency>
 <groupId>org.springframework</groupId>
 <artifactId>spring-context</artifactId>
 <version>${spring-version}</version>
 </dependency>
 <dependency>
 <groupId>org.springframework</groupId>
 <artifactId>spring-jms</artifactId>
 <version>${spring-version}</version>
 </dependency>
 <dependency>
 <groupId>org.springframework</groupId>
 <artifactId>spring-test</artifactId>
 <version>${spring-version}</version>
 </dependency>
 <dependency>
 <groupId>javax.annotation</groupId>
 <artifactId>jsr250-api</artifactId>
 <version>1.0</version>
 </dependency>
 <dependency>
 <groupId>org.apache.activemq</groupId>
 <artifactId>activemq-core</artifactId>
 <version>5.7.0</version>
 </dependency>
 </dependencies>

### **1.2.1 activeMQ准备** ###

既然是使用的*apache*的*activeMQ*作为*JMS*的实现，那么首先我们应该到*apache*官网上下载*activeMQ*（*[http://activemq.apache.org/download.html][http_activemq.apache.org_download.html]*），进行解压后运行其*bin*目录下面的*activemq.bat*文件启动*activeMQ*。

### **1.2.2配置ConnectionFactory** ###

*ConnectionFactory*是用于产生到*JMS*服务器的链接的，*Spring*为我们提供了多个*ConnectionFactory*，有*SingleConnectionFactory*和*CachingConnectionFactory*。*SingleConnectionFactory*对于建立*JMS*服务器链接的请求会一直返回同一个链接，并且会忽略*Connection*的*close*方法调用。*CachingConnectionFactory*继承了*SingleConnectionFactory*，所以它拥有*SingleConnectionFactory*的所有功能，同时它还新增了缓存功能，它可以缓存*Session*、*MessageProducer*和*MessageConsumer*。这里我们使用*SingleConnectionFactory*来作为示例。

这样就定义好产生*JMS*服务器链接的*ConnectionFactory*了吗？答案是非也。*Spring*提供的*ConnectionFactory*只是*Spring*用于管理*ConnectionFactory*的，真正产生到*JMS*服务器链接的*ConnectionFactory*还得是由*JMS*服务厂商提供，并且需要把它注入到*Spring*提供的*ConnectionFactory*中。我们这里使用的是*ActiveMQ*实现的*JMS*，所以在我们这里真正的可以产生*Connection*的就应该是由*ActiveMQ*提供的*ConnectionFactory*。所以定义一个*ConnectionFactory*的完整代码应该如下所示：

<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
 <property name="brokerURL" value="tcp://localhost:61616"/>
 </bean>
 
 
 <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 
 <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
 </bean>

*ActiveMQ*为我们提供了一个*PooledConnectionFactory*，通过往里面注入一个*ActiveMQConnectionFactory*可以用来将*Connection*、*Session*和*MessageProducer*池化，这样可以大大的减少我们的资源消耗。当使用*PooledConnectionFactory*时，我们在定义一个*ConnectionFactory*时应该是如下定义：

<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
 <property name="brokerURL" value="tcp://localhost:61616"/>
 </bean>
 
 <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
 <property name="connectionFactory" ref="targetConnectionFactory"/>
 <property name="maxConnections" value="10"/>
 </bean>
 
 <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>
 </bean>

### **1.2.3配置生产者** ###

配置好*ConnectionFactory*之后我们就需要配置生产者。生产者负责产生消息并发送到*JMS*服务器，这通常对应的是我们的一个业务逻辑服务实现类。但是我们的服务实现类是怎么进行消息的发送的呢？这通常是利用*Spring*为我们提供的*JmsTemplate*类来实现的，所以配置生产者其实最核心的就是配置进行消息发送的*JmsTemplate*。对于消息发送者而言，它在发送消息的时候要知道自己该往哪里发，为此，我们在定义*JmsTemplate*的时候需要往里面注入一个*Spring*提供的*ConnectionFactory*对象。

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
 
 <property name="connectionFactory" ref="connectionFactory"/>
 </bean>

在真正利用*JmsTemplate*进行消息发送的时候，我们需要知道消息发送的目的地，即*destination*。在*Jms*中有一个用来表示目的地的*Destination*接口，它里面没有任何方法定义，只是用来做一个标识而已。当我们在使用*JmsTemplate*进行消息发送时没有指定*destination*的时候将使用默认的*Destination*。默认*Destination*可以通过在定义*jmsTemplate bean*对象时通过属性*defaultDestination*或*defaultDestinationName*来进行注入，*defaultDestinationName*对应的就是一个普通字符串。在*ActiveMQ*中实现了两种类型的*Destination*，一个是点对点的*ActiveMQQueue*，另一个就是支持订阅*/*发布模式的*ActiveMQTopic*。在定义这两种类型的*Destination*时我们都可以通过一个*name*属性来进行构造，如：

<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
 <constructor-arg>
 <value>queue</value>
 </constructor-arg>
 </bean>
 
 <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
 <constructor-arg value="topic"/>
 </bean>

假设我们定义了一个*ProducerService*，里面有一个向*Destination*发送纯文本消息的方法*sendMessage*，那么我们的代码就大概是这个样子：

package com.tiantian.springintejms.service.impl;
     
    import javax.annotation.Resource;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
     
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Component;
     
    import com.tiantian.springintejms.service.ProducerService;
     
    @Component
    public class ProducerServiceImpl implements ProducerService {
     
        private JmsTemplate jmsTemplate;
        
        public void sendMessage(Destination destination, final String message) {
            System.out.println("---------------生产者发送消息-----------------");
            System.out.println("---------------生产者发了一个消息：" + message);
            jmsTemplate.send(destination, new MessageCreator() {
                public Message createMessage(Session session) throws JMSException {
                    return session.createTextMessage(message);
                }
            });
        } 
     
        public JmsTemplate getJmsTemplate() {
            returnjmsTemplate;
        } 
     
        @Resource
        public void setJmsTemplate(JmsTemplate jmsTemplate) {
            this.jmsTemplate = jmsTemplate;
        }
     
    }

我们可以看到在*sendMessage*方法体里面我们是通过*jmsTemplate*来发送消息到对应的*Destination*的。到此，我们生成一个简单的文本消息并把它发送到指定目的地*Destination*的生产者就配置好了。

### **1.2.4配置消费者** ###

生产者往指定目的地*Destination*发送消息后，接下来就是消费者对指定目的地的消息进行消费了。那么消费者是如何知道有生产者发送消息到指定目的地*Destination*了呢？这是通过*Spring*为我们封装的消息监听容器*MessageListenerContainer*实现的，它负责接收信息，并把接收到的信息分发给真正的*MessageListener*进行处理。每个消费者对应每个目的地都需要有对应的*MessageListenerContainer*。对于消息监听容器而言，除了要知道监听哪个目的地之外，还需要知道到哪里去监听，也就是说它还需要知道去监听哪个*JMS*服务器，这是通过在配置*MessageConnectionFactory*的时候往里面注入一个*ConnectionFactory*来实现的。所以我们在配置一个*MessageListenerContainer*的时候有三个属性必须指定，一个是表示从哪里监听的*ConnectionFactory*；一个是表示监听什么的*Destination*；一个是接收到消息以后进行消息处理的*MessageListener*。*Spring*一共为我们提供了两种类型的*MessageListenerContainer*，*SimpleMessageListenerContainer*和*DefaultMessageListenerContainer*。

*SimpleMessageListenerContainer*会在一开始的时候就创建一个会话*session*和消费者*Consumer*，并且会使用标准的*JMS MessageConsumer.setMessageListener()*方法注册监听器让*JMS*提供者调用监听器的回调函数。它不会动态的适应运行时需要和参与外部的事务管理。兼容性方面，它非常接近于独立的*JMS*规范，但一般不兼容*Java EE*的*JMS*限制。

大多数情况下我们还是使用的*DefaultMessageListenerContainer*，跟*SimpleMessageListenerContainer*相比，*DefaultMessageListenerContainer*会动态的适应运行时需要，并且能够参与外部的事务管理。它很好的平衡了对*JMS*提供者要求低、先进功能如事务参与和兼容*Java EE*环境。

**定义处理消息的*****MessageListener***

要定义处理消息的*MessageListener*我们只需要实现*JMS*规范中的*MessageListener*接口就可以了。*MessageListener*接口中只有一个方法*onMessage*方法，当接收到消息的时候会自动调用该方法。

package com.tiantian.springintejms.listener;
     
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
     
    public class ConsumerMessageListener implements MessageListener {
     
        public void onMessage(Message message) {
            //这里我们知道生产者发送的就是一个纯文本消息，所以这里可以直接进行强制转换
            TextMessage textMsg = (TextMessage) message;
            System.out.println("接收到一个纯文本消息。");
            try {
                System.out.println("消息内容是：" + textMsg.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
     
    }

有了*MessageListener*之后我们就可以在*Spring*的配置文件中配置一个消息监听容器了。

<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
 <constructor-arg>
 <value>queue</value>
 </constructor-arg>
 </bean>
 
 <bean id="consumerMessageListener" class="com.tiantian.springintejms.listener.ConsumerMessageListener"/> 
 
 
 <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
 <property name="connectionFactory" ref="connectionFactory" />
 <property name="destination" ref="queueDestination" />
 <property name="messageListener" ref="consumerMessageListener" />
 </bean>

我们可以看到我们定义了一个名叫*queue*的*ActiveMQQueue*目的地，我们的监听器就是监听了发送到这个目的地的消息。

至此我们的生成者和消费者都配置完成了，这也就意味着我们的整合已经完成了。这个时候完整的*Spring*的配置文件应该是这样的：

<?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
 xmlns:jms="http://www.springframework.org/schema/jms"
 xsi:schemaLocation="http://www.springframework.org/schema/beans
 http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
 http://www.springframework.org/schema/context
 http://www.springframework.org/schema/context/spring-context-3.0.xsd
 http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
 http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">
 
 <context:component-scan base-package="com.tiantian" />
 
 
 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
 
 <property name="connectionFactory" ref="connectionFactory"/>
 </bean>
 
 
 <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
 <property name="brokerURL" value="tcp://localhost:61616"/>
 </bean>
 
 
 <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 
 <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
 </bean>
 
 
 <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
 <constructor-arg>
 <value>queue</value>
 </constructor-arg>
 </bean>
 
 <bean id="consumerMessageListener" class="com.tiantian.springintejms.listener.ConsumerMessageListener"/>
 
 <bean id="jmsContainer"
 class="org.springframework.jms.listener.DefaultMessageListenerContainer">
 <property name="connectionFactory" ref="connectionFactory" />
 <property name="destination" ref="queueDestination" />
 <property name="messageListener" ref="consumerMessageListener" />
 </bean>
 </beans>

接着我们来测试一下，看看我们的整合是否真的成功了，测试代码如下：

package com.tiantian.springintejms.test;
 
 import javax.jms.Destination;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import com.tiantian.springintejms.service.ProducerService;
 
 @RunWith(SpringJUnit4ClassRunner.class)
 @ContextConfiguration("/applicationContext.xml")
 public class ProducerConsumerTest {
 
 @Autowired
 private ProducerService producerService;
 @Autowired
 @Qualifier("queueDestination")
 private Destination destination;
 
 @Test
 public void testSend() {
 for (int i=0; i<2; i++) {
 producerService.sendMessage(destination, "你好，生产者！这是消息：" + (i+1));
 }
 }
 
 }

在上面的测试代码中我们利用生产者发送了两个消息，正常来说，消费者应该可以接收到这两个消息。运行测试代码后控制台输出如下：

![bc449d33-84f8-3742-bca5-0251132e810f.png][]

看，控制台已经进行了正确的输出，这说明我们的整合确实是已经成功了。

[http_activemq.apache.org_download.html]: http://activemq.apache.org/download.html
[bc449d33-84f8-3742-bca5-0251132e810f.png]: /images/20220226/8ad64da780de499ea5b700ac4b69fb7f.png