ActiveMQ和spring整合，订阅多个topic并实现监听器监听

深藏阁楼爱情的钟 2022-05-19 14:19 84阅读 0赞

本文章适用初学ActiveMQ的同学。

本演示为windows下进行

首先下载ActiveMQ

1.ActiveMQ地址 ：http://activemq.apache.org/

2.下载完成后解压到本地； 这里我是解压在H:\\ruanjian\\activeMQ 盘

![70][]

3.启动本地ActiveMQ服务。

进入到安装目录双击activemq.bat文件启动，如果启动时窗口一闪而过，可以在cmd窗口，进入按章目录 执行  activemq.bat start 进行启动。

![70 1][]

![70 2][]

看到该窗口说明已启动成功，不要关闭此窗口。

接下来就改开发我们的代码了

首先在工程中添加需要的jar包

<dependency>    
         <groupId>org.apache.activemq</groupId>    
         <artifactId>activemq-core</artifactId>  
         <version>5.7.0</version>  
    </dependency>   
    <dependency>    
         <groupId>org.apache.activemq</groupId>  
         <artifactId>activemq-pool</artifactId>  
         <version>5.12.1</version>    
    </dependency>  
      
    <dependency>    
         <groupId>org.springframework</groupId>    
         <artifactId>spring-jms</artifactId>  
         <version>4.1.3.RELEASE</version>  
    </dependency>

这里我们创建两个工程，一个是消息生产者，一个是消息消费者

在我们真实的项目中，我们可能会有这样的需求：

1：在商品添加成功后，需要更新商品的索引；

2：会员注册成功后，向用户发送注册成功邮件；

这两个需求我们都可以通过ActiveMQ来实现，我们发布一个主题，让邮件系统和索引系统订阅这个主题，当邮件系统和索引系统监测有消息来时，我们就可以执行我们具体的业务了。

下来我们介绍具体的配置

我们先看消息生产者的配置：

springmvc的配置这里就不做具体的介绍了，我们详细介绍ActiveMQ的配置文件

spring-mq-active.xml：

<?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:amq="http://activemq.apache.org/schema/core"
    xmlns:jms="http://www.springframework.org/schema/jms" xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="
            http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd
            http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.1.xsd
            http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.12.1.xsd">
     
     
    
    <bean id="activeMqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://localhost:61616"/>
    <property name="userName" value="admin"></property>
    <property name="password" value="admin"></property>
    </bean>
     
    
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    <constructor-arg ref="activeMqConnectionFactory" />
    <property name="sessionCacheSize" value="100" />
    </bean>
     
     
    
    <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="pubSubDomain" value="false" />
    </bean>
     
    
    <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="pubSubDomain" value="true" />
    
    
    <property name="explicitQosEnabled" value="true"></property>
    </bean>
    </beans>

一、首先我们要配置连接就是ConnectionFactory；

这里的ConnectionFactory分为JMS厂商提供的ConnectionFactory和spring管理服务厂商ConnectionFactory的ConnectionFactory；

比较绕口，就是说spring要管理JSM厂商的ConnectionFactory；

说明：上边代码中的activeMqConnectionFactory就是ActiveMQ提供的ConnectionFactory，connectionFactory就是spring管理ConnectionFactory 的 ConnectionFactory；

Spring提供的ConnectionFactory只是Spring用于管理ConnectionFactory的，真正产生到JMS服务器链接的ConnectionFactory还得是由JMS服务厂商提供，并且需要把它注入到Spring提供的ConnectionFactory中。我们这里使用的是ActiveMQ实现的JMS，所以在我们这里真正的可以产生Connection的就应该是由ActiveMQ提供的ConnectionFactory

spring提供了三中管理器：

1>、SingleConnectionFactory：对于建立JMS服务器链接的请求会一直返回同一个链接，并且会忽略Connection的close方法调用。  
2>、CachingConnectionFactory：继承了SingleConnectionFactory，所以它拥有SingleConnectionFactory的所有功能，同时它还新增了缓存功能，  
      它可以缓存Session、MessageProducer和MessageConsumer。我们使用CachingConnectionFactory来作为示例。  
3>、PooledConnectionFactory：线程池

配置如下

<bean id="activeMqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    	<property name="brokerURL" value="tcp://localhost:61616"/>
    	<property name="userName" value="admin"></property>
    	<property name="password" value="admin"></property>
    </bean>
    	
    
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    	<constructor-arg ref="activeMqConnectionFactory" />
    	<property name="sessionCacheSize" value="100" />
    </bean>

二、消息发送模板

连接配置完成后，下来就要配置消息发送模板，既然是发消息我们就要知道是谁发消息，怎么发消息；

connectionFactory是负责生产消息并发送至JMS服务器；但是具体怎么发，就需要我们的spring提供的JmsTemplate类来实现。所以我们配置消息生产者的核心就是配置JmsTemplate。对于发消息来说，首先是要连接服务器，发什么类型的消息，往哪发。

消息类型：ActiveMQ提供了点对点的队列模式和主题(发布/订阅)模式；

<bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
    <property name="connectionFactory" ref="connectionFactory" />
    <property name="pubSubDomain" value="true" />
    
    
    <property name="explicitQosEnabled" value="true"></property>
    </bean>
    </beans>

到此消生产者就已经配置完成了。

三、消费者配置

所谓消费者就是，消息生产者将消息发送到了JMS服务器，那么就得有消费者处理消费这条信息。消息生产者将消息发送至指定的目的地，那么消费者就要开始消费。

那么消费者是怎样知道生产者给自己发消息了呢？所以每个消费者都应有一个对应的消息监听器MessageListenerContainer。而对于监听器来说它要知道去哪监听，监听哪个消息目的地，所以我们配置向监听既要知道 ConnectionFactory(从哪里监听) 和 destination(监听什么)，监听到消息后怎么处理，这三个核心的东西。

<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">  
    <constructor-arg value="goodsAddTopic,goods2AddTopic" />  
    </bean>  
      
      
    <bean id="goodsAddMessageListener" class="com.sxdax.mq.consumer.message.GoodsESMessageLister"/>  
      
      
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
    <property name="connectionFactory" ref="connectionFactory"></property>  
    <property name="destination" ref="topicDestination"></property>  
    <property name="messageListener" ref="goodsAddMessageListener"></property>
    </bean>

connectionFactory这里就不说了，配置同上变的一样。

消费者端ActiveMQ配置文件spring-mq-active.xml

<?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:amq="http://activemq.apache.org/schema/core"
    xmlns:jms="http://www.springframework.org/schema/jms" xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="
            http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd
            http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.1.xsd
            http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.12.1.xsd">
     
     
    
    <bean id="activeMqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://localhost:61616"/>
    <property name="userName" value="admin"></property>
    <property name="password" value="admin"></property>
    </bean>
     
    
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
    <constructor-arg ref="activeMqConnectionFactory" />
    <property name="sessionCacheSize" value="100" />
    </bean>
     
     
    
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg value="goodsAddTopic,goods2AddTopic" />
    </bean>
     
    
    <bean id="goodsAddMessageListener" class="com.sxdax.mq.consumer.message.GoodsESMessageLister"/>
     
    
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory"></property>
    <property name="destination" ref="topicDestination"></property>
    <property name="messageListener" ref="goodsAddMessageListener"></property>
    </bean>
    </beans>

生产者控制层

@Controller
    @RequestMapping("test")
    public class TestController {
    	
    	@Autowired
    	TopicSender topicSender;
    	
    	@RequestMapping("send")
    	@ResponseBody
    	public String sendMessage(){
    		System.out.println("开始发送消息");
    		topicSender.send("goodsAddTopic", "商品添加成功,开始更新商品索引库");
    		return "test success";
    	}
    	
    	@RequestMapping("send2")
    	@ResponseBody
    	public String sendMessage2(){
    		System.out.println("开始发送消息-------------2");
    		topicSender.send("goods2AddTopic", "商品添加成功,开始更新商品索引库---------2");
    		return "test success";
    	}
    	
    }

生产者实现层：

@Component
    public class TopicSender {
    
    	@Autowired
    	@Qualifier("jmsTopicTemplate")
    	private JmsTemplate jmsTemplate;
    	
    	/**
    	 * 发送一条消息到指定的队列（目标）
    	 * @param queueName 队列名称
    	 * @param message 消息内容
    	 */
    	public void send(String topicName,final String message){
    		jmsTemplate.send(topicName, new MessageCreator() {
    			@Override
    			public Message createMessage(Session session) throws JMSException {
    				return session.createTextMessage(message);
    			}
    		});
    	}
    	
    }

注：TopicSender 类必须要扫描，不扫描会报空指针

消费者监听类：

/**
     * 消息监听器，消息的消费者
     * @author back
     */
    public class Goods2ESMessageLister implements SessionAwareMessageListener<Message>{
    
    
    	@Override
    	public void onMessage(Message message, Session arg1) throws JMSException {
    		try {
                Destination destination = message.getJMSDestination();
                String topic = destination.toString();
                System.out.println("topic====="+topic);
                TextMessage text=(TextMessage) message;
                System.out.println("消息是========"+text.getText());
            }catch (Exception e){
    
            }
    		
    	}
    
    }

在运行源代码之前，请先启动您本地的ActiveMQ服务。

[70]: /images/20220519/70c6ad2ef6564f69b64c3860d0c924f0.png
[70 1]: /images/20220519/68582edd39b54398b79d0fb120d831d1.png
[70 2]: /images/20220519/68a4550b5f874ae29f1ec379c0f40078.png