spring 整合 ActiveMQ

2022-10-17 09:52:37

1.1 JMS简介

JMS即Java Message Service，Java消息服务。主要用于在生产者和消费者之间进行消息传递，生产者负责产生消息，而消费者负责接收消息。

点对点：一个生产者对应一个消费者；

发布/订阅模式，一个生产者对应多个消费者。

1.2 Spring整合JMS

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准备

下载activeMQ（http://activemq.apache.org/download.html），解压运行bin目录activemq.bat启动activeMQ。

1.2.2 配置ConnectionFactory

ConnectionFactory：SingleConnectionFactory和CachingConnectionFactory。

SingleConnectionFactory：单连接，忽略Connection的close方法调用。

CachingConnectionFactory：继承SingleConnectionFactory，新增缓存功能，可以缓存Session、MessageProducer和MessageConsumer。

 <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"/>

配置：

     <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->

     <bean id="mqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

         <property name="brokerURL" value="tcp://localhost:61616"/>

     </bean>

     <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->

     <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

         <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->

         <property name="targetConnectionFactory" ref="mqConnectionFactory"/>

     </bean>

当使用PooledConnectionFactory连接池：

     <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->

     <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 配置生产者

     <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->

     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

         <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->

         <property name="connectionFactory" ref="connectionFactory"/>

     </bean>

队列：

     <!--这个是队列目的地，点对点的-->

     <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>

生产者：

 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;

     }

 }

1.2.4 配置消费者

定义处理消息的MessageListener：实现JMS规范中的MessageListener接口

 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();

         }

     }

 }

消息监听容器配置：

     <!--这个是队列目的地-->

     <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>

完整配置：

 <?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" />

     <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->

     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

         <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->

         <property name="connectionFactory" ref="connectionFactory"/>

     </bean>

     <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->

     <bean id="mqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

         <property name="brokerURL" value="tcp://localhost:61616"/>

     </bean>

     <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->

     <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

         <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->

         <property name="targetConnectionFactory" ref="mqConnectionFactory"/>

     </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));

         }

     }

 }

码农公寓