activeMQ(二)--spring整合activeMQ

忘是亡心i 2022-05-29 04:56 234阅读 0赞

1.  \#\#\# 工程结构

![这里写图片描述][SouthEast]

1.  ### pom ###
    
        <dependencies>
         <dependency>
           <groupId>junit</groupId>
           <artifactId>junit</artifactId>
           <version>4.12</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-test</artifactId>
           <version>${spring-version}</version>
           <scope>test</scope>
         </dependency>
        
         <dependency>
           <groupId>org.apache.activemq</groupId>
           <artifactId>activemq-all</artifactId>
           <version>5.13.2</version>
         </dependency>
        
         <dependency>
           <groupId>org.springframework</groupId>
           <artifactId>spring-jms</artifactId>
         </dependency>
        
         <dependency>
           <groupId>org.springframework</groupId>
           <artifactId>spring-messaging</artifactId>
           <version>${spring-version}</version>
         </dependency>
        </dependencies>
2.  ### 配置文件 ###
    
    activemq.xml
    
        <context:component-scan base-package="com.susq" />
        
        <context:property-placeholder location="config.properties" />
        
        
        <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
         <property name="brokerURL" value="${activemq.url}"/>
        </bean>
        
        
        <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
         
         <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
        </bean>
        
        
        <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
         <constructor-arg ref="connectionFactory" />
         
         <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="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
         <constructor-arg>
           
           <value>${activemq.queue}</value>
         </constructor-arg>
        </bean>
        
        
        <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
         <constructor-arg value="${activemq.topic}"/>
        </bean>
    
    config.properties
    
        activemq.url=tcp://localhost:61616
        activemq.queue=su.queue
        activemq.topic=su.topic
    
    log4j.properties
    
        log4j.rootLogger=ERROR,console
        log4j.appender.console=org.apache.log4j.ConsoleAppender
        log4j.appender.console.layout=org.apache.log4j.PatternLayout
        log4j.appender.console.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss,SSS} [%c]-[%p] %m%n
        
        
        #给自己项目设置日志级别
        
        log4j.logger.com.susq=DEBUG
    
    
3.  ### 工程代码 ###
    
    JmsTemplate 是spring 提供的简化了同步JMS访问代码的帮助类。文档：[https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/jms/core/JmsTemplate.html][https_docs.spring.io_spring_docs_current_javadoc-api_org_springframework_jms_core_JmsTemplate.html]
    
    生产者
    
        package com.susq.service.impl;
        
        import com.susq.service.ProducerService;
        import lombok.extern.slf4j.Slf4j;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.beans.factory.annotation.Qualifier;
        import org.springframework.jms.core.JmsTemplate;
        import org.springframework.stereotype.Component;
        
        import javax.jms.Destination;
        import javax.jms.Session;
        
        @Component("producerService")
        @Slf4j
        public class ProducerServiceImpl implements ProducerService { 
        
        @Autowired
        @Qualifier("jmsQueueTemplate")
        private JmsTemplate jmsTemplate;
        
        public void sendMessage(Destination destination, final String message) {
            log.debug("---------------生产者发送消息-----------------");
            log.debug("---------------生产者发了一个消息：" + message);
            /* send()方法第二个参数是MessageCreator类型,它里面只有一个方法，所以他是一个函数式接口，支持lambda表达式 public interface MessageCreator { Message createMessage(Session var1) throws JMSException; } 为了防止“函数是接口” 变成"非函数接口”，我们可以在这个上面加上一个声明@FunctionalInterface， 这样别人就无法在里面添加新的接口函数了。 */
            jmsTemplate.send(destination, (Session session) -> session.createTextMessage(message));
        }
        }
    
    消费者
    
        package com.susq.service.impl;
        
        import com.susq.service.ConsumerService;
        import lombok.extern.slf4j.Slf4j;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.beans.factory.annotation.Qualifier;
        import org.springframework.jms.core.JmsTemplate;
        import org.springframework.stereotype.Component;
        
        import javax.jms.Destination;
        import javax.jms.JMSException;
        import javax.jms.TextMessage;
        
        @Slf4j
        @Component("consumerService")
        public class ConsumerServiceImpl implements ConsumerService { 
        
           @Autowired
           @Qualifier("jmsQueueTemplate")
           private JmsTemplate jmsTemplate;
        
           public void receiveMessage(Destination destination) {
               log.debug("------------消费消息--------------");
               while (true) {
                   try {
                       //使用JMSTemplate接收消息
                       TextMessage txtmsg = (TextMessage) jmsTemplate.receive(destination);
                       if (null != txtmsg) {
                           log.debug("--- 收到消息内容为: " + txtmsg.getText());
                       } else {
                           break;
                       }
                   } catch (JMSException e) {
                       log.error("错误 {}", e);
                   }
               }
           }
        }
    
    测试
    
        package com.susq;
        
        import com.susq.service.ConsumerService;
        import com.susq.service.ProducerService;
        import org.junit.Test;
        import org.junit.runner.RunWith;
        import org.springframework.beans.factory.annotation.Autowired;
        import org.springframework.test.context.ContextConfiguration;
        import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
        
        import javax.jms.Destination;
        
        /** * Unit test for simple App. */
        @RunWith(SpringJUnit4ClassRunner.class)
        @ContextConfiguration(locations = "classpath:activemq.xml")
        public class AppTest { 
        
           @Autowired
           private ProducerService producerService;
        
           @Autowired
           private Destination queueDestination;
        
           @Autowired
           private ConsumerService consumerService;
        
           @Test
           public void testProducer() {
               producerService.sendMessage(queueDestination, "生产消息");
           }
        
           @Test
           public void testConsumer() {
               consumerService.receiveMessage(queueDestination);
           }
        
        }
    
    运行生产者，以后，可以看到发送了一条消息等待消费，对ActiveMQ自带Web管控台不熟悉的，可以参考上一篇 [http://blog.csdn.net/u013041642/article/details/79547128][http_blog.csdn.net_u013041642_article_details_79547128]
    
    ![这里写图片描述][SouthEast 1]
    
    运行消费者
    
    ![这里写图片描述][SouthEast 2]

消费者中我们写了一个死循环，使线程持续运行，接受消息。如果不循环，只运行一次，那每次只能消费一条消息。

### 5. 消息监听器 ###

上面我们是用 jmsTemplate.receive(destination) 来接受消息。而使用消息监听器是一种更优雅的消息接受方式，只需要实现消息监听器接口MessageListener即可。topic监听器写法也一样。

import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /** * @author susq * @since 2018-03-15-21:18 */
    @Slf4j
    @Component("consumerServiceNew")
    public class ConsumerServiceNew implements MessageListener { 
    
        @Override
        public void onMessage(Message message) {
            try {
                log.debug("ConsumerServiceNew接收到消息："+((TextMessage)message).getText());
            } catch (JMSException e) {
                log.debug("接收消息异常，{}", e);
            }
        }
    }

activemq.xml中增加监听容器的配置, 需要设置监听的地址和监听器的bean.

<jms:listener-container>
            <jms:listener destination="${activemq.queue}" ref="consumerServiceNew"/>
        </jms:listener-container>
    
        
        <jms:listener-container destination-type="topic" >
            <jms:listener destination="${activemq.topic}" ref="topicConsumer"/>
        </jms:listener-container>

测试代码

package com.susq;
    
    import com.susq.service.ConsumerService;
    import com.susq.service.ProducerService;
    import com.susq.service.TopicConsumer;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    import javax.jms.Destination;
    
    /** * Unit test for simple App. */
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations = "classpath:activemq.xml")
    public class AppTest { 
    
        @Autowired
        private ProducerService producerService;
    
        @Autowired
        private Destination queueDestination;
    
        @Autowired
        private Destination topicDestination;
    
        @Test
        public void testProducer() {
            producerService.sendMessage(queueDestination, "queue生产消息");
        }
    
        @Test
        public void testTopicProducer() {
            producerService.sendMessage(topicDestination, "topic生产消息");
        }
    }

运行生产者，加载配置文件的时候，我们配好的监听容器同时被加载，所以生产消息和监听消息同时进行。

todo: 队列消息的监听器，加载以后可以一次性接受之前生产的未消费的所有队列消息。但是主题消息的监听器，不能收到监听器加载之前的消息。有待研究。

消息监听器除了MessageListener还有SessionAwareMessageListener和MessageListenerAdapter，后面再总结。

[SouthEast]: /images/20220529/d43f75be55a34fe1a816ce1a1f12fa53.png
[https_docs.spring.io_spring_docs_current_javadoc-api_org_springframework_jms_core_JmsTemplate.html]: https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/jms/core/JmsTemplate.html
[http_blog.csdn.net_u013041642_article_details_79547128]: http://blog.csdn.net/u013041642/article/details/79547128
[SouthEast 1]: /images/20220529/570b7d0d49d641c4b6c8712a64efc3ce.png
[SouthEast 2]: /images/20220529/496a2ae5e4a3475ab9585e1e20f417fe.png