ActiveMQ与Spring的整合使用

在介绍ActiveMQ特性的时候有一条就是，对spring的支持，ActiveMQ可以很容易内嵌到使用Spring的系统里面去，而且也支持Spring2.0的特性，那么这里就来进行二者的整合与测试使用。
1、生产者
创建maven工程spring-activemq-producer，整一个spring的配置文件放到resource目录下。
在配置文件applicationContext.xml中
1、配置ConnectionFactory；
2、配置spring提供的用于管理ConnectionFactory的ConnectionFactory；
3、配置生成者，Spring为我们提供了JsmTemplate这个工具类，用它可以进行消息发送；
4、配置目的地（有点对点的和一对多的）；

<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
    <bean id="targetConnectionFactory" 
        class="org.apache.activemq.spring.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.25.128:61616"/> 
    </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 class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>
    <!--这个是队列目的地，点对点的，仅有两个构造函数一个无参一个带String-->
    <bean id="queue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="spring.activemq.queue"/>
    </bean>
    <!--这个是主题目的地，一对多的,仅有两个构造函数一个无参一个带String -->
    <bean id="topic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="spring.activemq.topic"/>
    </bean>

编写测试类，测试点对点消息发送

public class ActiveMQTest {

    /*
     * 测试点对点：生产者
     */
    @Test
    public void test(){
        ApplicationContext ac = new ClassPathXmlApplicationContext("classpath:applicationContext-activemq.xml");
        //获取生产者
        JmsTemplate jmsTemplate = ac.getBean(JmsTemplate.class);
        //获取消息发送目的地,由于配置了两个目的地所以需要根据id来取，这里测试的是点对点的，测试发布/订阅要使用id为"topic"
        Destination queue = (Destination) ac.getBean("queue");
        //发送消息，需要传入生产者
        jmsTemplate.send(queue,new MessageCreator() {

            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage message = session.createTextMessage("发送了一个消息");
                return message;
            }
        });
    }
}

运行方法后页面查看如下

image.png image.png

发现一个挂起的消息（该消息是被持久化的，被消费时消失），一个入队消息。

2、消费者
创建另外一个maven工程spring-activemq-consumer，整一个spring的配置文件放到resource目录下。
在配置文件applicationContext.xml中
1、配置ConnectionFactory；
2、配置spring提供的用于管理ConnectionFactory的ConnectionFactory；
3、配置目的地（有点对点的和一对多的），到这都是跟配置生产者一致；
4、编写消息监听器，MyMessageListener（名字随意）实现MessageListener接口。

public class MyMessageListener implements MessageListener{

    @Override
    public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage)message;
        try {
            //获取消息信息内容
            String text = textMessage.getText();
            //打印信息
            System.out.println(text);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

5、配置文件中配置自定义消息监听器；
6、配置消息监听容器。

<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供 -->
    <bean id="targetConnectionFactory" 
        class="org.apache.activemq.spring.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.25.128:61616"/> 
    </bean>
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>

    <!--这个是队列目的地，点对点的，仅有两个构造函数一个无参一个带String-->
    <bean id="queue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="Spring.activemq.queue"/>
    </bean>
    <!--这个是主题目的地，一对多的,仅有两个构造函数一个无参一个带String -->
    <bean id="topic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="Spring.activemq.topic"/>
    </bean>
    <!-- 配置监听器 -->
    <bean id="messageListener" class="com.itheima.activemq.MyMessageListener"></bean>
    <!-- 消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <!--这里测试是点对点的，使用发布/订阅的话使用topic-->
        <property name="destination" ref="queue"/>
        <property name="messageListener" ref="messageListener"/>
    </bean>

注意：消费者接收消息目的地必须跟生产者发送消息目的地名称一致，不然不在一个频道是接收不到消息的。
测试接收消息，由于前面发送了一个消息，这里是应该能接收到的。

public class ConsumerTest {
    @Test
    public void testConsumer() throws IOException{
        ApplicationContext ac = 
                new ClassPathXmlApplicationContext("classpath:applicationContext-activemq.xml"); 
        System.in.read();
    }
}

注：System.in.read()表示按下回车程序才会继续往下执行，否则是处于一直能接收消息的状态，生成者发送几条消息这就接收几条（也就会打印几条生产者发送的消息的内容 “发送了一个消息“）。

控制台输出：
log4j:WARN No appenders could be found for logger (org.springframework.core.env.StandardEnvironment).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
发送了一个消息

查看页面：

image.png

能看到被挂起的消息被消费了，所以有一条出队消息，一个消费者。这里测试的是点对点的，关于发布/订阅形式的上面改一下配置文件，以及测试方法中发布消息目的地改为”topic”即可，注释里面已经写过就不再写测试了。