Spring整合ActiveMQ

前言

上一篇文章JMS的独立使用，看到了整个消息的发送与接收过程非常简单，但是其中却掺杂着大量的冗余代码，比如Connection的创建于关闭，Session的创建于关闭等等，为了消除这一冗余工作量，Spring进行了进一步的封装。

Spring整合ActiveMQ

（1）Spring的配置文件

配置文件是Spring的核心，Spring整合消息服务的使用也从配置文件配置开始。类似于数据库操作，Spring也将ActiveMQ中的操作统一封装至JmsTemplate中，以方便我们统一使用。所以，在Spring的核心配置文件中首先要注册JmsTemplate类型的bean。当然，ActiveMQConnectionFactory用于连接消息服务器，是消息服务的基础，也要注册。ActiveMQQueue则用于指定消息的目的地。

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL">
            <value>tcp://localhost:61616</value>
        </property>
    </bean>

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>

    <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg index="0" value="HelloWordQueue"/>
    </bean>

 （2）发送端

有了以上的配置，Spring就可以根据配置信息简化我们的工作量。Spring中使用发送消息到消息服务器，省去了冗余的connection和session等的创建与销毁等工作。

public class HelloWordSender {

    public static void main(String[] args){
        ApplicationContext context = new ClassPathXmlApplicationContext("spring.xml");
        JmsTemplate jmsTemplate = (JmsTemplate) context.getBean("jmsTemplate");
        Destination destination = (Destination) context.getBean("destination");
        jmsTemplate.send(destination, new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return session.createTextMessage("Hello, this is a test!");
            }
        });
    }
}

（3）接收端

public class HelloWordReciver {
    public static void main(String[] args) throws JMSException {
        ApplicationContext context = new ClassPathXmlApplicationContext("spring.xml");
        JmsTemplate jmsTemplate = (JmsTemplate) context.getBean("jmsTemplate");
        Destination destination = (Destination) context.getBean("destination");

        TextMessage message = (TextMessage) jmsTemplate.receive(destination);
        System.out.println("reviced message is :" + message.getText());
    }
}

到这里我们已经完成了Spring的消息的发送与接收操作。

先启动服务端代码，再启动客户端代码：

客户端：

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.
reviced message is :Hello, this is a test!

发现，收到后客户端立即关闭。

所以，如HelloWordReciver中所示的代码，使用jmsTemplate.receive(destination)方法只能接收一次消息，如果未接收到消息，则会一直等待，当然用户可以设置timeout等属性来控制等待时间，但是一旦接收到消息本次接收任务就会结束，虽然用户可以通过while(true)的方式来实现循环监听消息服务器上的消息，还有一种更好的方法：创建消息监听器。

（4）创建消息监听器

用于监听消息，一旦有新的消息Spring会将消息引导至消息监听器以方便用户进行相应的逻辑处理。

public class MyMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        TextMessage msg = (TextMessage) message;
        try {
            System.out.println(msg.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

（5）修改配置文件

在配置文件中加入：

<bean id="myTextListener" class="com.joe.mytag.activemq.MyMessageListener"/>

    <bean id="javaConsumer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="destination"/>
        <property name="messageListener" ref="myTextListener"/>
    </bean>

通过以上的修改便可以进行消息监听的功能了，一旦有消息传入至消息服务器，则会被消息监听器监听到，并由Spring将消息内容引导至消息监听器的处理函数中等待用户的进一步逻辑处理。

参考：《Spring源码深度解析》 郝佳编著：