Terry.Li-彬

簡介

實戰(zhàn)一 , 實戰(zhàn)二介紹了ActiveMQ的基本概念和配置方式.

本篇將通過一個實例介紹使用spring發(fā)送,消費topic, queue類型消息的方法. 不懂topic和queue的google 之.

如圖示, TOPIC和QUEUE分別代表一個topic和一個queue消息通道.

1. TopicMessageProducer向topic發(fā)送消息, TopicConsumerA和TopicConsumerB則從topic消費消息.
2. QueueMessageProducer向Queue發(fā)送消息, QueueConsumer從Queue中消費消息

Spring整合JMS

就像對orm, web的支持一樣, spring同樣支持jms, 為整合jms到已有的項目提供了很多便利的方法. 本篇主要講實戰(zhàn), 是所以先從配置開始, spring配置jms基本上需要8個部分.

1. ConnectionFactory. 和jms服務(wù)器的連接, 可以是外部的jms server, 也可以使用embedded ActiveMQ Broker.
2. Destination. 有topic和queue兩種方式.
3. JmsTemplate. spring提供的jms模板.
4. MessageConverter. 消息轉(zhuǎn)換器.
5. MessageProducer. 消息生產(chǎn)者.
6. MessageConsumer. 消息消費者.
7. MessageListener. 消息監(jiān)聽器
8. MessageListenerContainer. 消息監(jiān)聽容器

下面以實例的方式介紹上面8個部分.

1. ConnectionFactory

Xml代碼

1. <amq:connectionFactory id="jmsConnectionFactory" brokerURL="vm://localhost" />  

<amq:connectionFactory id="jmsConnectionFactory" brokerURL="vm://localhost" />

 brokerURL是指要連接的activeMQ server的地址, activeMQ提供了多種brokerURL, 集體可參見文檔.一般我們使用嵌套的ActiveMQ server. 配置如下, 這個配置使用消息的存儲機制, 服務(wù)器重啟也不會丟失消息.

Xml代碼

1. <!--  embedded ActiveMQ Broker -->  
2.     <amq:broker useJmx="false" persistent="true">  
3.         <amq:persistenceAdapter>  
4.             <amq:amqPersistenceAdapter directory="d:/amq"/>  
5.         </amq:persistenceAdapter>  
6.         <amq:transportConnectors>  
7.             <amq:transportConnector uri="tcp://localhost:61616" />  
8.                        <amq:transportConnector uri="vm://localhost:0" />  
9.         </amq:transportConnectors>  
10.     </amq:broker>  

<!--  embedded ActiveMQ Broker -->
<amq:broker useJmx="false" persistent="true">
<amq:persistenceAdapter>
<amq:amqPersistenceAdapter directory="d:/amq"/>
</amq:persistenceAdapter>
<amq:transportConnectors>
<amq:transportConnector uri="tcp://localhost:61616" />
<amq:transportConnector uri="vm://localhost:0" />
</amq:transportConnectors>
</amq:broker>

 2. Destination

 在實例中我們使用了兩種destination

Xml代碼

1. <!--  ActiveMQ destinations  -->  
2. <!--  使用topic方式-->  
3. <amq:topic name="TOPIC" physicalName="JMS-TEST-TOPIC" />  
4. <!--  使用Queue方式-->  
5. <amq:queue name="QUEUE" physicalName="JMS-TEST-QUEUE" />  

<!--  ActiveMQ destinations  -->
<!--  使用topic方式-->
<amq:topic name="TOPIC" physicalName="JMS-TEST-TOPIC" />
<!--  使用Queue方式-->
<amq:queue name="QUEUE" physicalName="JMS-TEST-QUEUE" />

 3. JmsTemplate

Xml代碼

1. <!--  Spring JmsTemplate config -->  
2.     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
3.         <property name="connectionFactory">  
4.             <!--  lets wrap in a pool to avoid creating a connection per send -->  
5.             <bean class="org.springframework.jms.connection.SingleConnectionFactory">  
6.                 <property name="targetConnectionFactory" ref="jmsConnectionFactory" />  
7.             </bean>  
8.         </property>  
9.         <!-- custom MessageConverter -->  
10.         <property name="messageConverter" ref="defaultMessageConverter" />  
11.     </bean>  

<!--  Spring JmsTemplate config -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<!--  lets wrap in a pool to avoid creating a connection per send -->
<bean class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="jmsConnectionFactory" />
</bean>
</property>
<!-- custom MessageConverter -->
<property name="messageConverter" ref="defaultMessageConverter" />
</bean>

  4. MessageConverter

   MessageConverter實現(xiàn)的是org.springframework.jms.support.converter.MessageConverter接口, 提供消息的轉(zhuǎn)換功能. DefaultMessageConverter的實現(xiàn)見附件.

Xml代碼

1. <bean id="defaultMessageConverter" class="com.andyao.activemq.DefaultMessageConverter" />  

<bean id="defaultMessageConverter" class="com.andyao.activemq.DefaultMessageConverter" />

  5. MessageProducer

   實例擁有兩個消息生產(chǎn)者, 消息生產(chǎn)者都是POJO, 實現(xiàn)見附件.

Xml代碼

1. <!-- POJO which send Message uses  Spring JmsTemplate -->  
2.     <bean id="topicMessageProducer" class="com.andyao.activemq.TopicMessageProducer">  
3.         <property name="template" ref="jmsTemplate" />  
4.         <property name="destination" ref="TOPIC" />  
5.     </bean>  
6.     <bean id="queueMessageProducer" class="com.andyao.activemq.QueuMessageProducer">  
7.         <property name="template" ref="jmsTemplate" />  
8.         <property name="destination" ref="QUEUE" />  
9.     </bean>  

<!-- POJO which send Message uses  Spring JmsTemplate -->
<bean id="topicMessageProducer" class="com.andyao.activemq.TopicMessageProducer">
<property name="template" ref="jmsTemplate" />
<property name="destination" ref="TOPIC" />
</bean>
<bean id="queueMessageProducer" class="com.andyao.activemq.QueuMessageProducer">
<property name="template" ref="jmsTemplate" />
<property name="destination" ref="QUEUE" />
</bean>

 6. MessageConsumer

 TOPIC通道有兩個消息消費者, QUEUE有一個消息消費者

Xml代碼

1. <!--  Message Driven POJO (MDP) -->  
2.     <!-- consumer1 for topic a -->  
3.     <bean id="topicConsumerA" class="com.andyao.activemq.TopicConsumerA" />  
4.     <!-- consumer2 for topic a -->  
5.     <bean id="topicConsumerB" class="com.andyao.activemq.TopicConsumerB" />  
6.     <!-- consumer for queue -->  
7.     <bean id="queueConsumer" class="com.andyao.activemq.QueueConsumer" />  

<!--  Message Driven POJO (MDP) -->
<!-- consumer1 for topic a -->
<bean id="topicConsumerA" class="com.andyao.activemq.TopicConsumerA" />
<!-- consumer2 for topic a -->
<bean id="topicConsumerB" class="com.andyao.activemq.TopicConsumerB" />
<!-- consumer for queue -->
<bean id="queueConsumer" class="com.andyao.activemq.QueueConsumer" />

  7. MessageListener

每一個消息消費者都對應(yīng)一個MessageListener

Xml代碼

1. <bean id="topicListenerA" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
2.         <constructor-arg ref="topicConsumerA" />  
3.         <!--  may be other method -->  
4.         <property name="defaultListenerMethod" value="receive" />  
5.         <!-- custom MessageConverter define -->  
6.         <property name="messageConverter" ref="defaultMessageConverter" />  
7.     </bean>  
8.   
9.     <bean id="topicListenerB" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
10.         <constructor-arg ref="topicConsumerB" />  
11.         <!--  may be other method -->  
12.         <property name="defaultListenerMethod" value="receive" />  
13.         <!-- custom MessageConverter define -->  
14.         <property name="messageConverter" ref="defaultMessageConverter" />  
15.     </bean>  
16.   
17.     <bean id="queueListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
18.         <constructor-arg ref="queueConsumer" />  
19.         <!--  may be other method -->  
20.         <property name="defaultListenerMethod" value="receive" />  
21.         <!-- custom MessageConverter define -->  
22.         <property name="messageConverter" ref="defaultMessageConverter" />  
23.     </bean>  

<bean id="topicListenerA" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicConsumerA" />
<!--  may be other method -->
<property name="defaultListenerMethod" value="receive" />
<!-- custom MessageConverter define -->
<property name="messageConverter" ref="defaultMessageConverter" />
</bean>
<bean id="topicListenerB" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicConsumerB" />
<!--  may be other method -->
<property name="defaultListenerMethod" value="receive" />
<!-- custom MessageConverter define -->
<property name="messageConverter" ref="defaultMessageConverter" />
</bean>
<bean id="queueListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="queueConsumer" />
<!--  may be other method -->
<property name="defaultListenerMethod" value="receive" />
<!-- custom MessageConverter define -->
<property name="messageConverter" ref="defaultMessageConverter" />
</bean>

 8. MessageListenerContainer

 有幾個MessageListener既有幾個MessageListenerContainer

Xml代碼

1. <bean id="topicListenerContainerA" class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
2.         <property name="connectionFactory" ref="jmsConnectionFactory" />  
3.         <property name="destination" ref="TOPIC" />  
4.         <property name="messageListener" ref="topicListenerA" />  
5.     </bean>  
6.   
7.     <bean id="topicListenerContainerB" class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
8.         <property name="connectionFactory" ref="jmsConnectionFactory" />  
9.         <property name="destination" ref="TOPIC" />  
10.         <property name="messageListener" ref="topicListenerB" />  
11.     </bean>  
12.        
13.     <bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
14.         <property name="connectionFactory" ref="jmsConnectionFactory" />  
15.         <property name="destination" ref="QUEUE" />  
16.         <property name="messageListener" ref="queueListener" />  
17.     </bean>  

<bean id="topicListenerContainerA" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="destination" ref="TOPIC" />
<property name="messageListener" ref="topicListenerA" />
</bean>
<bean id="topicListenerContainerB" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="destination" ref="TOPIC" />
<property name="messageListener" ref="topicListenerB" />
</bean>
<bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="destination" ref="QUEUE" />
<property name="messageListener" ref="queueListener" />
</bean>

  Summary

寫spring配置文件的時候, 要把MessageProducer, MessageConsumer,MessageListener,MessageListenerContainer幾個地方弄清楚:

1. 可以有一個或者多個消息生產(chǎn)者向同一個destination發(fā)送消息.
2. queue類型的只能有一個消息消費者.
3. topic類型的可以有多個消息消費者.
4. 每個消費者對應(yīng)一個MessageListener和一個MessageListenerContainer.