ActiveMQ与Spring集成.

介绍

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。由于ActiveMQ是一个独立的JMS provider，很容易被用于多种结构设计中，也不需要第三方服务器的支持，这样我们的应用可以部署在各种J2EE服务器或servlet服务器中，也可以独立运行。

 

本文介绍如何在spring中集成activemq，然后演示下如何发送和接收消息。

 

在Spring中集成ActiveMQ

配置ConnectionFactory

在spring中设置ActiveMQ的connectionFactory，需要导入activemq-client.jar，可以到activemq官网或者通过maven repository下载。

<!-- 真正可以产生Connection的ConnectionFactory，由对应的JMS服务厂商提供--> 

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

        <property name="brokerURL" value="tcp://localhost:61616" />

</bean>

 

配置一个消息队列Queue（点对点）

<!--这个是队列目的地，点对点的-->

<bean id="notifyQueue" class="org.apache.activemq.command.ActiveMQQueue">

     <constructor-arg>

         <value>notifyQueue</value>

     </constructor-arg>

</bean>

 

发送消息配置

配置JmsTemplate

使用Spring提供的JmsTemplate封装JMS的细节，简化代码开发，JmsTemplate支持发送和接收。

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

    <property name="connectionFactory" ref="connectionFactory"/>

    <property name="receiveTimeout" value="2000" />

</bean>

 

JmsTemplate每次发送都会创建新的connect，新的session，新的MessageProducter，这样效率较低。Spring提供了一个org.springframework.jms.connection.SingleConnectionFactory，每次都返回同一个connection，它会忽略close，所以connection会一直保持着。Spring还提供了一个org.springframework.jms.connection.CachingConnectionFactory，

增加了对session的缓存功能，可以通过sessionCacheSize设置，但因为它继承自SingleConnectionFactory，所以同样是一直使用同一connection，我觉得主要是用于测试环境。

 

ActiveMQ也提供了一个对connection,session, MessageProducer实现池化功能的org.apache.activemq.pool.PooledConnectionFactory。这个可以在生产环节使用。不过他和atomikos的分布式事务目前测试配合不了，使用atomikos分布式事务时由其他方法，下回再说。

编写消息生产者服务

编写消息生产者，把消息发送至队列notifyQueue

@Component

publicclass JmsProducerService {

   

    @Resource

    private JmsTemplate jmsTemplate;

   

    @Qualifier("notifyQueue")

    @Autowired

    private Destination destination;

   

    publicvoid sendMessage(final String message) { 

        System.out.println("---------------生产者-----------------");

        System.out.println("发了一个消息：" + message); 

        jmsTemplate.send(destination, new MessageCreator() {

           

            @Override

            public Message createMessage(Session session) throws JMSException {

                returnsession.createTextMessage(message); 

            } 

        });

      

    }

}

 

接收消息配置

配置监听容器

配置一个Spring监听容器，引用上面配置好的connectionFactory

<!-- 配置一个通用的JMS监听容器 -->

<bean id="jmsListenerContainerFactory"

 class="org.springframework.jms.config.DefaultJmsListenerContainerFactory">

     <property name="connectionFactory" ref="connectionFactory"/>

</bean>

 

配置@JmsListener注解支持

使用@Jmslistener直接在代码中定义监听器，减少配置文件编写，需要添加下面配置以支持@Jmslistener

<!-- 支持@JmsListener自动启动监听器 -->

<jms:annotation-driven/>

 

PS：<jms:annotation-driven>默认是使用jmsListenerContainerFactory，所以配置中可以不用指定

 

编写消息监听服务

使用@JmsListener定义一个监听器，监听消息队列notifyQueue

@Component

publicclass AnConsumerListener {

 

    @JmsListener(destination="notifyQueue")

    public void receiveMessage(String message){

       System.out.println("---------------消费者监听-----------------");

       System.out.println("接收到一个纯文本消息：" + message);

      

    }

}

 

测试

 

启动ActiveMQ

执行activemq/bin/activemq.bat start

 

使用生产者发生一条消息看看

JmsProducerService jms = (JmsProducerService)SpringContextManager.getBean("jmsProducerService");

jms.sendMessage("Hello world!");