activeMQ(二)--spring整合activeMQ-蒲公英云

### 工程结构

这里写图片描述

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>

配置文件

activemq.xml

<context:component-scan base-package="com.susq" />
<context:property-placeholder location="config.properties" />
<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
 <property name="brokerURL" value="${activemq.url}"/>
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
 <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>
<!-- Spring提供的JMS工具类，它可以进行消息发送、接收等，设置它为队列模式 -->
<bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
 <constructor-arg ref="connectionFactory" />
 <!-- 非pub/sub模型（发布/订阅），即队列模式 -->
 <property name="pubSubDomain" value="false" />
</bean>
<!-- 设置它为Topic类型 -->
<bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
 <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
 <constructor-arg ref="connectionFactory" />
 <!-- pub/sub模型（发布/订阅） -->
 <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 HHss,SSS} [%c]-[%p] %m%n

    #给自己项目设置日志级别
    log4j.logger.com.susq=DEBUG

工程代码

JmsTemplate 是spring 提供的简化了同步JMS访问代码的帮助类。文档：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

这里写图片描述

运行消费者

这里写图片描述

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

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.

<!-- 定义Queue监听器 -->
    <jms:listener-container>
        <jms:listener destination="${activemq.queue}" ref="consumerServiceNew"/>
    </jms:listener-container>
    <!-- 定义topic监听器 -->
    <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生产消息");
    }
}