ActiveMQ实战(六)--Spring整合ActiveMQ实现队列和主题发布订阅通信

偏执的太偏执、 2022-06-04 07:54 203阅读 0赞

**一、新建Maven项目**

工具使用的是IDEA，具体操作如下，New-->Project

![20171208115144002][]

![20171208101109939][]

![20171208101222838][]

![20171208101232800][]

创建的项目如下所示

![20171208120954870][]

然后在main下新建java目录、创建好之后、在java目录上右键转换为source

![20171208135855284][]

完整的项目结构如下:

![20171208211604745][]

**二、添加依赖**

POM.XML文件如下:

<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.activemq</groupId>
      <artifactId>activemq_demo</artifactId>
      <packaging>war</packaging>
      <version>1.0-SNAPSHOT</version>
      <name>activemq_demo Maven Webapp</name>
      <url>http://maven.apache.org</url>
    
      <developers>
        <developer>
          <id>fendo</id>
          <name>fendo</name>
          <email>fendo@qq.com</email>
        </developer>
      </developers>
    
    
      <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <fastjson.vesrion>1.2.35</fastjson.vesrion>
        <activemq.version>5.13.2</activemq.version>
        <activemq-pool.version>5.15.0</activemq-pool.version>
        <spring.version>4.1.9.RELEASE</spring.version>
        <junit.version>4.12</junit.version>
      </properties>
    
      <dependencies>
    
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-core</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-oxm</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-tx</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-jdbc</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-aop</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-context</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-context-support</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-expression</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-orm</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-web</artifactId>
          <version>${spring.version}</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-webmvc</artifactId>
          <version>${spring.version}</version>
        </dependency>
    
        <dependency>
          <groupId>org.apache.xbean</groupId>
          <artifactId>xbean-spring</artifactId>
          <version>4.4</version>
        </dependency>
    
        <dependency>
          <groupId>commons-logging</groupId>
          <artifactId>commons-logging</artifactId>
          <version>1.2</version>
        </dependency>
    
        <dependency>
          <groupId>com.alibaba</groupId>
          <artifactId>fastjson</artifactId>
          <version>${fastjson.vesrion}</version>
        </dependency>
        
        <dependency>
          <groupId>org.apache.activemq</groupId>
          <artifactId>activemq-all</artifactId>
          <version>${activemq.version}</version>
        </dependency>
        <dependency>
          <groupId>org.apache.activemq</groupId>
          <artifactId>activemq-pool</artifactId>
          <version>${activemq-pool.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>
          <scope>test</scope>
        </dependency>
        <dependency>
          <groupId>junit</groupId>
          <artifactId>junit</artifactId>
          <version>${junit.version}</version>
          <scope>test</scope>
        </dependency>
      </dependencies>
    
      <build>
        <finalName>activemq_demo</finalName>
      </build>
    </project>

**三、添加配置文件**

**web.xml**

<?xml version="1.0" encoding="UTF-8"?>
    <web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://java.sun.com/xml/ns/javaee" 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>
      <context-param>
        <param-name>log4jExposeWebAppRoot</param-name>
        <param-value>false</param-value>
      </context-param>
      
    
      
      <filter>
        <filter-name>encodingFilter</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>encodingFilter</filter-name>
        <url-pattern>/*</url-pattern>
      </filter-mapping>
      
    
      
      <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>classpath*:/spring-context*.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>
      
    </web-app>

**spring-context.xml**

spring-mvc.xml

spring-contexe-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.13.2.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">
            
            <property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
            
            
            
            <property name="sessionCacheSize" value="100" />
        </bean>
    
    
        
    
    
        
    
        
        <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <constructor-arg ref="connectionFactory"></constructor-arg>
            
            <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>
        
    
        
    
        
        <bean id="QueueConsumer_1"
              class="com.fendo.activemq.service.consumer.QueueConsumer_1"></bean>
        <bean id="QueueConsumer_2"
              class="com.fendo.activemq.service.consumer.QueueConsumer_2"></bean>
    
        
        <bean id="TopicConsumer_1"
              class="com.fendo.activemq.service.consumer.TopicConsumer_1"></bean>
        <bean id="TopicConsumer_2"
              class="com.fendo.activemq.service.consumer.TopicConsumer_2"></bean>
    
    
        
    
        
        <jms:listener-container destination-type="queue"
                                container-type="default" connection-factory="connectionFactory"
                                acknowledge="auto">
            <jms:listener destination="queue" ref="QueueConsumer_1" />
            <jms:listener destination="queue" ref="QueueConsumer_2" />
        </jms:listener-container>
        
        <jms:listener-container destination-type="topic"
                                container-type="default" connection-factory="connectionFactory"
                                acknowledge="auto">
            <jms:listener destination="topic" ref="TopicConsumer_1" />
            <jms:listener destination="topic" ref="TopicConsumer_2" />
        </jms:listener-container>
        
    
        
        <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</value>
            </constructor-arg>
        </bean>
    
    </beans>

log4j.properties

### direct log messages to stdout and logFile###
    log4j.rootCategory=INFO, stdout,logFile
    
    # OpenSymphony Stuff
    log4j.logger.com.opensymphony=INFO
    log4j.logger.org.apache.commons=INFO 
    
    # Spring Stuff
    log4j.logger.org.springframework=INFO
    log4j.logger.org.springframework.oxm=INFO
    
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=[\u65F6\u95F4\:%d{yyyy-MM-dd hh\:mm\:ss}] [\u7EA7\u522B\:%p] [\u7C7B\:%c]  [\u6D88\u606F\:%m] %n 
    
    log4j.appender.logFile=org.apache.log4j.RollingFileAppender
    log4j.appender.logFile.File=D\:\\demo.log
    log4j.appender.logFile.layout=org.apache.log4j.PatternLayout
    log4j.appender.logFile.layout.ConversionPattern=[\u65F6\u95F4\:%d{yyyy-MM-dd hh\:mm\:ss}] [\u7EA7\u522B\:%p] [\u7C7B\:%c]  [\u6D88\u606F\:%m] %n 
    log4j.appender.logFile.MaxFileSize = 5MB
    log4j.appender.logFile.MaxBackupIndex =3

**四、添加生产者消费者**

**4.1、Queue队列通信**

新建消息生产者:

package com.fendo.activemq.service.producer;
    
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Service;
    
    /**
     * @version V1.0
     * @Author fendo
     * @ClassName ProducerService
     * @PackageName com.fendo.activemq.service
     * @Description 队列-消息生产者 Service
     * @Data 2017-09-21 10:47
     **/
    @Service
    public class QueueProducerService {
    
        @Autowired
        private JmsTemplate jmsQueueTemplate;
        //发送消息
        public void sendMessage(Destination destination,final String message) {
            System.out.println("QueueProducerService发送消息："+message);
            jmsQueueTemplate.send(destination, new MessageCreator() {
    
                @Override
                public Message createMessage(Session session) throws JMSException {
                    return session.createTextMessage(message);
                }
            });
        }
    }

然后创建多个消息消费者

**消费者1：**

package com.fendo.activemq.service.consumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * @version V1.0
     * @Author fendo
     * @ClassName QueueConsumer_1
     * @PackageName com.fendo.activemq.listener
     * @Description 队列消费者1
     * @Data 2017-09-21 10:47
     **/
    public class QueueConsumer_1 implements MessageListener {
    
        //当收到消息后，自动调用该方法
        @Override
        public void onMessage(Message message) {
            TextMessage tm = (TextMessage) message;
            try {
                System.out.println("QueueConsumer_1接收到消息内容是：" + tm.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

**消费者2：**

package com.fendo.activemq.service.consumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * @version V1.0
     * @Author fendo
     * @ClassName QueueConsumer_2
     * @PackageName com.fendo.activemq.listener
     * @Description 队列消费者2
     * @Data 2017-09-21 10:47
     **/
    public class QueueConsumer_2 implements MessageListener {
    
        //当收到消息后，自动调用该方法
        @Override
        public void onMessage(Message message) {
            TextMessage tm = (TextMessage) message;
            try {
                System.out.println("QueueConsumer_2接收到消息内容是：" + tm.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

**4.2、Topic通信**

创建Topic生产者:

package com.fendo.activemq.service.producer;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Service;
    
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    /**
     * @version V1.0
     * @Author fendo
     * @ClassName TopicConsumerService
     * @PackageName com.fendo.activemq.service
     * @Description 主题-生产者 Service
     * @Data 2017-09-21 10:47
     **/
    @Service
    public class TopicProducerService {
    
        @Autowired
        private JmsTemplate jmsTopicTemplate;
    
        //发送消息
        public void sendMessage(Destination destination, final String message) {
            System.out.println("向主题" + destination.toString() + "发送了消息------------" + message);
            jmsTopicTemplate.send(destination, new MessageCreator() {
                @Override
                public Message createMessage(Session session) throws JMSException {
                    return session.createTextMessage(message);
                }
            });
        }
    }

然后创建多个Topic消费者:

package com.fendo.activemq.service.consumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * @version V1.0
     * @Author fendo
     * @ClassName TopicConsumer_1
     * @PackageName com.fendo.activemq.service
     * @Description 主题-消息消费者1
     * @Data 2017-09-21 10:47
     **/
    public class TopicConsumer_1 implements MessageListener {
    
        @Override
        public void onMessage(Message message) {
            TextMessage textMessage = (TextMessage) message;
            try {
                System.out.println("TopicConsumer_1接收到消息内容是：" + textMessage.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    
    }

package com.fendo.activemq.service.consumer;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * @version V1.0
     * @Author fendo
     * @ClassName TopicConsumer_2
     * @PackageName com.fendo.activemq.service
     * @Description 主题-消息消费者2
     * @Data 2017-09-21 10:47
     **/
    public class TopicConsumer_2 implements MessageListener {
    
        @Override
        public void onMessage(Message message) {
            TextMessage textMessage = (TextMessage) message;
            try {
                System.out.println("TopicConsumer_2接收到消息内容是：" + textMessage.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    
    }

Controller类:

package com.fendo.activemq.controller;
    
    import com.fendo.activemq.service.producer.QueueProducerService;
    import com.fendo.activemq.service.producer.TopicProducerService;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    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.RequestParam;
    import org.springframework.web.servlet.ModelAndView;
    
    import javax.annotation.Resource;
    import javax.jms.Destination;
    
    /**
     * @version V1.0
     * @Author fendo
     * @ClassName ActivemqController
     * @PackageName com.fendo.activemq.controller
     * @Description Activemq Controller
     * @Data 2017-09-21 10:47
     **/
    @Controller
    public class ActivemqController {
    
        private Logger logger = LoggerFactory.getLogger(ActivemqController.class);
    
    
        @Autowired
        private QueueProducerService queueProducerService;
    
        @Autowired
        private TopicProducerService topicProducerService;
    
        @Resource
        @Qualifier("queueDestination")
        private Destination queueDestination;
    
        @Resource
        @Qualifier("topicDestination")
        private Destination topicDestination;
    
        @RequestMapping("")
        public ModelAndView activemq(){
            ModelAndView mv = new ModelAndView();
            mv.setViewName("index");
            return mv;
        }
    
    
        /**
         * 去发消息页面
         * @return
         */
        @RequestMapping(value="/producer",method= RequestMethod.GET)
        public ModelAndView producer(){
            ModelAndView mv = new ModelAndView();
            mv.setViewName("producer");
            return mv;
        }
    
        /**
         * 发送消息
         * @param message
         * @return
         */
        @RequestMapping(value="/onsend",method=RequestMethod.POST)
        public ModelAndView producer(@RequestParam("message") String message,@RequestParam("sendflag") String sendflag) {
            System.out.println("------------send to jms------------");
            ModelAndView mv = new ModelAndView();
            //1 主题  2 队列
            if("1".equals(sendflag)){
                System.out.println("topic生产者产生消息：" + message);
                topicProducerService.sendMessage(topicDestination, "topic生产者产生消息：" + message);
            }else {
                System.out.println("队列生产者产生消息：" + message);
                queueProducerService.sendMessage(queueDestination, "队列生产者产生消息："  + message);
            }
            mv.setViewName("index");
            return mv;
        }
    
    }

**五、测试项目**

然后创建消息生产者页面:

<%@ page language="java" contentType="text/html; charset=UTF-8"
        pageEncoding="UTF-8"%>
    <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
    <html>
    <head>
    <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
    <title>生产者</title>
    </head>
    <body>
        <h1>生产者</h1>
    
        <h2>去队列发消息</h2>
        <form action="${pageContext.request.contextPath}/onsend" method="post">
            发送的内容:<textarea name="message" ></textarea>
            <input type="text" style="display: none" name="sendflag" value="2" />
            <input type="submit" value="提交" />
        </form>
    
        <h2>去主题发消息</h2>
        <form action="${pageContext.request.contextPath}/onsend" method="post">
            发送的内容:<textarea name="message" ></textarea>
            <input type="text" style="display: none" name="sendflag" value="1" />
            <input type="submit" value="提交" />
        </form>
    
        <h2><a href="${pageContext.request.contextPath}/">返回主页</a></h2>
    </body>
    </html>

效果如下:

![20171208210605458][]

可以分别发送队列和主题，发送的效果如下:

![20171208210704319][]

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

完整项目如下:[http://download.csdn.net/download/u011781521/10151680][http_download.csdn.net_download_u011781521_10151680]

[20171208115144002]: /images/20220604/f433dae284874293a750145100007714.png
[20171208101109939]: /images/20220604/ba4afe4eb6f24cb99d0d4e612709428f.png
[20171208101222838]: /images/20220604/0f0a57fd66524165aa74c9cb34a20818.png
[20171208101232800]: /images/20220604/fd052a35d1c24dc69a4ef9c23eaad531.png
[20171208120954870]: /images/20220604/b2520cc37a91400f898b3ed35d7851cf.png
[20171208135855284]: /images/20220604/8488180811634745b7f0008fc1d38485.png
[20171208211604745]: /images/20220604/e7cb9c84ebc94d86ae5874f3aee2261d.png
[20171208210605458]: /images/20220604/560168e418964cf08d29e85001288289.png
[20171208210704319]: /images/20220604/3a51afd505284e59bcd707d7939ab7aa.png
[http_download.csdn.net_download_u011781521_10151680]: http://download.csdn.net/download/u011781521/10151680