ActiveMQ实战(五)--Spring整合ActiveMQ实现点对点与主题发布订阅通信

太过爱你忘了你带给我的痛 2022-06-12 12:56 252阅读 0赞

**一、项目结构**

![20170723200427663][]

**二、加入Maven依赖**

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
      <modelVersion>4.0.0</modelVersion>
      <groupId>com.fendo</groupId>
      <artifactId>Spring_ActiveMQ</artifactId>
      <packaging>war</packaging>
      <version>0.0.1-SNAPSHOT</version>
      <name>Spring_ActiveMQ Maven Webapp</name>
      <url>http://maven.apache.org</url>
      
      
      <properties>
          <springframework>4.1.8.RELEASE</springframework>
          <javax.servlet>3.1.0</javax.servlet>
      </properties>
      
      
      <dependencies>
      
       <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>4.10</version>
          <scope>test</scope>
        </dependency>
        
        <dependency>
            <groupId>jstl</groupId>
            <artifactId>jstl</artifactId>
            <version>1.2</version>
        </dependency>
        
      
        <dependency>
    	    <groupId>javax.servlet</groupId>
    	    <artifactId>javax.servlet-api</artifactId>
    	    <version>${javax.servlet}</version>
        </dependency>
        
        
        <dependency>
    	    <groupId>log4j</groupId>
    	    <artifactId>log4j</artifactId>
    	    <version>1.2.16</version>
        </dependency>
        
        
        
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>${springframework}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>${springframework}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-tx</artifactId>
            <version>${springframework}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
            <version>${springframework}</version>
        </dependency>
        <dependency>  
            <groupId>org.springframework</groupId>  
            <artifactId>spring-jms</artifactId>  
            <version>${springframework}</version>  
        </dependency>
        
        
        
        <dependency>
            <groupId>org.apache.xbean</groupId>
            <artifactId>xbean-spring</artifactId>
            <version>3.16</version>
        </dependency>
        
        
        <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>
        
      </dependencies>
      
      
      <build>
        <finalName>Spring_ActiveMQ</finalName>
      </build>
    </project>

**三、配置文件**

applicationContext.xml

spring-mvc.xml

<?xml version="1.0" encoding="UTF-8"?>
    
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:aop="http://www.springframework.org/schema/aop"
           xmlns:context="http://www.springframework.org/schema/context"
           xmlns:mvc="http://www.springframework.org/schema/mvc"
           xmlns:tx="http://www.springframework.org/schema/tx"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
           xsi:schemaLocation="http://www.springframework.org/schema/aop
            http://www.springframework.org/schema/aop/spring-aop-4.0.xsd
            http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
            http://www.springframework.org/schema/context
            http://www.springframework.org/schema/context/spring-context-4.0.xsd
            http://www.springframework.org/schema/mvc
            http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd
            http://www.springframework.org/schema/tx
            http://www.springframework.org/schema/tx/spring-tx-4.0.xsd">
    
        
        <mvc:annotation-driven />
        
            
        <mvc:resources location="/resources/" mapping="/resources/**"/>
        
        
        <context:component-scan base-package="com" >
            
            <context:include-filter type="annotation" expression="org.springframework.stereotype.Controller"/>
        </context:component-scan>
    
        
        <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver">
            <property name="prefix" value="/" />
            <property name="suffix" value=".jsp" />
            
            <property name="order" value="1" />
        </bean>
    </beans>

ActiveMQ.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"
           xmlns:mvc="http://www.springframework.org/schema/mvc"
           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/mvc
            http://www.springframework.org/schema/mvc/spring-mvc-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.8.0.xsd">
    
        
        
        <amq:connectionFactory id="amqConnectionFactory"
                               brokerURL="tcp://127.0.0.1:61616"
                               userName="admin"
                               password="admin" />
    
        
        <bean id="connectionFactory"
              class="org.springframework.jms.connection.CachingConnectionFactory">
              
                 
            <constructor-arg ref="amqConnectionFactory" />
            
            
            <property name="sessionCacheSize" value="100" />
        </bean>
    
        
        <bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
            
            <constructor-arg>
                <value>fendo</value>
            </constructor-arg>
        </bean>
    
    
          
    
        
        
        <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
            <property name="connectionFactory" ref="connectionFactory" />
            <property name="defaultDestination" ref="demoQueueDestination" />
            <property name="receiveTimeout" value="10000" />
            
            
            
            <property name="pubSubDomain" value="false" />
        </bean>
    
    	  
    	<bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">  
    	         
    	    <constructor-arg ref="connectionFactory" />  
    	      
    	    <property name="pubSubDomain" value="true" />  
    	</bean>  
    	
    	  
    	
    	
          
    	
          
        <jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">  
            <jms:listener destination="test.queue" ref="queueReceiver1"/>  
            <jms:listener destination="test.queue" ref="queueReceiver2"/>  
        </jms:listener-container>  
      
          
        <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">  
            <jms:listener destination="test.topic" ref="topicReceiver1"/>  
            <jms:listener destination="test.topic" ref="topicReceiver2"/>  
        </jms:listener-container>  
    	
         
    	
    	
    
    	
    
    	    
    
    
    </beans>

web.xml

<web-app xmlns="http://java.sun.com/xml/ns/javaee"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
              http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
             version="3.0">
    
      <display-name>Archetype Created Web Application</display-name>
    
    
    	  
        <context-param>  
            <param-name>log4jConfigLocation</param-name>  
            <param-value>classpath:log4j.properties</param-value>  
        </context-param>  
        <context-param>  
            <param-name>log4jRefreshInterval</param-name>  
            <param-value>6000</param-value>  
        </context-param>  
          
        <listener>  
            <listener-class>org.springframework.web.util.Log4jConfigListener</listener-class>  
        </listener>  
          
     
     
      
      <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>
          classpath*:applicationContext.xml,classpath*:ActiveMQ.xml
        </param-value>
      </context-param>
    
      <listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
      </listener>
    
      <servlet>
        <servlet-name>springMVC</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
          <param-name>contextConfigLocation</param-name>
          <param-value>classpath:spring-mvc.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
      </servlet>
      <servlet-mapping>
        <servlet-name>springMVC</servlet-name>
        <url-pattern>/</url-pattern>
      </servlet-mapping>
    
      
      <filter>
        <filter-name>characterEncodingFilter</filter-name>
        <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
        <init-param>
          <param-name>encoding</param-name>
          <param-value>UTF-8</param-value>
        </init-param>
        <init-param>
          <param-name>forceEncoding</param-name>
          <param-value>true</param-value>
        </init-param>
      </filter>
      <filter-mapping>
        <filter-name>characterEncodingFilter</filter-name>
        <url-pattern>/*</url-pattern>
      </filter-mapping>
      
    </web-app>

**四、一些类**

两个Controller类,一个用来测试点对点通信,另一个是用来测试点对点与主题发布订阅通信:

点对点通信控制类:

package com.fendo.controller;
    
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    import com.fendo.service.ConsumerService;
    import com.fendo.service.ProducerService;
    
    import javax.annotation.Resource;
    import javax.jms.Destination;
    import javax.jms.TextMessage;
    
    /**
     * 
     * @author fendo
     *
     */
    @Controller
    public class MessageController {
    
        private Logger logger = LoggerFactory.getLogger(MessageController.class);
    
    	
    	@Resource(name = "demoQueueDestination")
        private Destination destination;
    
        //队列消息生产者
        @Resource(name = "producerService")
        private ProducerService producer;
    
        //队列消息消费者
        @Resource(name = "consumerService")
        private ConsumerService consumer;
    
        @RequestMapping(value = "/SendMessage", method = RequestMethod.POST)
        @ResponseBody
        public void send(String msg) {
            logger.info(Thread.currentThread().getName()+"------------send to jms Start");
            producer.sendMessage(msg);
            logger.info(Thread.currentThread().getName()+"------------send to jms End");
        }
    
        @RequestMapping(value= "/ReceiveMessage",method = RequestMethod.GET)
        @ResponseBody
        public Object receive(){
            logger.info(Thread.currentThread().getName()+"------------receive from jms Start");
            TextMessage tm = consumer.receive(destination);
            logger.info(Thread.currentThread().getName()+"------------receive from jms End");
            return tm;
        }
    
    }

主题发布订阅通信:

package com.fendo.controller;
    
    
    import javax.annotation.Resource;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    import com.fendo.producer.queue.QueueSender;
    import com.fendo.producer.topic.TopicSender;
    
    
    
    /**
     * 
     * @author fendo
     * @description controller测试
     */
    @Controller
    @RequestMapping("/activemq")
    public class ActivemqController {
    	
    	@Autowired 
    	QueueSender queueSender;
    	
    	
    	@Autowired 
    	TopicSender topicSender;
    	
    	/**
    	 * 发送消息到队列
    	 * Queue队列：仅有一个订阅者会收到消息，消息一旦被处理就不会存在队列中
    	 * @param message
    	 * @return String
    	 */
    	@ResponseBody
    	@RequestMapping("queueSender")
    	public String queueSender(@RequestParam("message")String message){
    		String opt="";
    		try {
    			queueSender.send("test.queue", message);
    			opt = "suc";
    		} catch (Exception e) {
    			opt = e.getCause().toString();
    		}
    		return opt;
    	}
    	
    	/**
    	 * 发送消息到主题
    	 * Topic主题 ：放入一个消息，所有订阅者都会收到 
    	 * 这个是主题目的地是一对多的
    	 * @param message
    	 * @return String
    	 */
    	@ResponseBody
    	@RequestMapping("topicSender")
    	public String topicSender(@RequestParam("message")String message){
    		String opt = "";
    		try {
    			topicSender.send("test.topic", message);
    			opt = "suc";
    		} catch (Exception e) {
    			opt = e.getCause().toString();
    		}
    		return opt;
    	}
    	
    }

一个消费者里面包含queue和topic

queue包下有QueueReceiver1和QueueReceiver2

QueueReceiver1

package com.fendo.consumer.queue;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Component;
    
    /**
     * 
     * @author fendo
     * @description  队列消息监听器
     * 
     */
    @Component
    public class QueueReceiver1 implements MessageListener {
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			System.out.println("QueueReceiver1接收到消息:"+((TextMessage)message).getText());
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    }

QueueReceiver2

package com.fendo.consumer.queue;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Component;
    
    /**
     * 
     * @author fendo
     * @description  队列消息监听器
     * 
     */
    @Component
    public class QueueReceiver2 implements MessageListener {
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			System.out.println("QueueReceiver2接收到消息:"+((TextMessage)message).getText());
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    
    }

topic包下有TopicReceiver1,TopicReceiver2

TopicReceiver1

package com.fendo.consumer.topic;
    
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Component;
    
    
    /**
     * 
     * @author fendo
     * @description  Topic消息监听器
     * 
     */
    @Component
    public class TopicReceiver1 implements MessageListener{
    
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			System.out.println("TopicReceiver1接收到消息:"+((TextMessage)message).getText());
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    	
    }

TopicReceiver2

package com.fendo.consumer.topic;
    
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import org.springframework.stereotype.Component;
    
    /**
     * 
     * @author fendo
     * @description  Topic消息监听器
     * 
     */
    @Component
    public class TopicReceiver2 implements MessageListener{
    
    
    	@Override
    	public void onMessage(Message message) {
    		try {
    			System.out.println("TopicReceiver2接收到消息:"+((TextMessage)message).getText());
    		} catch (JMSException e) {
    			e.printStackTrace();
    		}
    	}
    	
    }

**五、测试**

**A.点对点通信**

1）手动测试点对点通信

首先运行项目,然后访问: [ http://localhost:8080/Spring\_ActiveMQ/sendMessage.jsp][http_localhost_8080_Spring_ActiveMQ_sendMessage.jsp]

![20170723201807217][]

输入要发送的内容,点击发送,控制台打印如下

![20170723202054538][]

ActiveMQ控制台会多出一条记录

![20170723202141646][]

然后访问地址:http://localhost:8080/Spring\_ActiveMQ/ReceiveMessage.jsp 接受消息

![20170723202241333][]

点击"接受消息",控制台会打印出生产者发送的消息

![20170723202344526][]

2）自动监听点对点通信

在实际项目中，我们很少会自己手动去获取消息，而是使用一个监听器来监听是否有消息到达，这样子可以很快的完成对消息的处理。

在ActiveMQ.xml配置文件中有段代码是用来配置监听器的

<bean id="queueMessageListener" class="com.fendo.filter.QueueMessageListener" />
    
        
        <bean id="queueListenerContainer"
              class="org.springframework.jms.listener.DefaultMessageListenerContainer">
            <property name="connectionFactory" ref="connectionFactory" />
            <property name="destination" ref="demoQueueDestination" />
            <property name="messageListener" ref="queueMessageListener" />
        </bean>

QueueMessageListener类为

package com.fendo.filter;
    
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * 
     * @author fendo
     *
     */
    public class QueueMessageListener implements MessageListener {
        public void onMessage(Message message) {
            TextMessage tm = (TextMessage) message;
            try {
                System.out.println("QueueMessageListener监听到了文本消息：\t"
                        + tm.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

这样的话，我们在发消息的时候，后台就会自动的监听

![20170723203041668][]

注意:当开启自动监听以后,手动访问就会失效。

**B.主题发布订阅通信**

当访问: [ http://localhost:8080/Spring\_ActiveMQ/index.jsp][http_localhost_8080_Spring_ActiveMQ_index.jsp]

![20170723203226545][]

输入要发送的内容，然后点击发送Queue消息,会发现后台只有一个接受到了信息

![20170723203350722][]

然后发送Topic消息,后台两个都收到了信息

![20170723203525779][]

从上面可看出队列模型和PUB/SUB模型的区别，Queue只能由一个消费者接收，其他Queue中的成员无法接受到被已消费的信息，而Topic则可以，只要是订阅了Topic的消费者，全部可以获取到生产者发布的信息。

完整示例: [http://download.csdn.net/detail/u011781521/9908153][http_download.csdn.net_detail_u011781521_9908153]

[20170723200427663]: /images/20220612/c768de894e2142f6ac4a2e02cf8326c7.png
[http_localhost_8080_Spring_ActiveMQ_sendMessage.jsp]: http://localhost:8080/Spring_ActiveMQ/sendMessage.jsp
[20170723201807217]: /images/20220612/d8508f0f55cc4050a342a9db8c259687.png
[20170723202054538]: /images/20220612/279f7f7157494aa1aaf7e611de6a4d55.png
[20170723202141646]: /images/20220612/694aa2daf0a54954a68c14571b048b54.png
[20170723202241333]: /images/20220612/4f47bac5b18444338d5cd6a8ad39bd23.png
[20170723202344526]: /images/20220612/3c0cfd3c08724965a2fdaf956beda97b.png
[20170723203041668]: /images/20220612/40e53004f6d44a31bca282b6fbc8711e.png
[http_localhost_8080_Spring_ActiveMQ_index.jsp]: http://localhost:8080/Spring_ActiveMQ/index.jsp
[20170723203226545]: /images/20220612/d6a5dbc6cb6242a9a7904a2e55832404.png
[20170723203350722]: /images/20220612/d8364b29c60a4ec6a3d385eb8f45d589.png
[20170723203525779]: /images/20220612/7d325d2508e74ee19d81f385eb5eadba.png
[http_download.csdn.net_detail_u011781521_9908153]: http://download.csdn.net/detail/u011781521/9908153