簡介

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

本篇將通過一個例項介紹使用spring傳送,消費topic, queue型別訊息的方法. 不懂topic和queue的google 之.

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

1. TopicMessageProducer向topic傳送訊息, TopicConsumerA和TopicConsumerB則從topic消費訊息.
2. QueueMessageProducer向Queue傳送訊息, QueueConsumer從Queue中消費訊息

Spring整合JMS

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

1. ConnectionFactory. 和jms伺服器的連線, 可以是外部的jms server, 也可以使用embedded ActiveMQ Broker.
2. Destination. 有topic和queue兩種方式.
3. JmsTemplate. spring提供的jms模板.
4. MessageConverter. 訊息轉換器.
5. MessageProducer. 訊息生產者.
6. MessageConsumer. 訊息消費者.
7. MessageListener. 訊息監聽器
8. MessageListenerContainer. 訊息監聽容器

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

1. ConnectionFactory

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

 brokerURL是指要連線的activeMQ server的地址, activeMQ提供了多種brokerURL, 集體可參見文件.一般我們使用巢狀的ActiveMQ server. 配置如下, 這個配置使用訊息的儲存機制, 伺服器重啟也不會丟失訊息.

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>  

 2. Destination

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

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" />  

 3. JmsTemplate

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>  

  4. MessageConverter

   MessageConverter實現的是org.springframework.jms.support.converter.MessageConverter介面, 提供訊息的轉換功能. DefaultMessageConverter的實現見附件.

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

  5. MessageProducer

   例項擁有兩個訊息生產者, 訊息生產者都是POJO, 實現見附件.

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>  

 6. MessageConsumer

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

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" />  

  7. MessageListener

每一個訊息消費者都對應一個MessageListener

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.     <bean id="topicListenerB" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
9.         <constructor-arg ref="topicConsumerB" />  
10.         <!--  may be other method -->  
11.         <property name="defaultListenerMethod" value="receive" />  
12.         <!-- custom MessageConverter define -->  
13.         <property name="messageConverter" ref="defaultMessageConverter" />  
14.     </bean>  
15.     <bean id="queueListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">  
16.         <constructor-arg ref="queueConsumer" />  
17.         <!--  may be other method -->  
18.         <property name="defaultListenerMethod" value="receive" />  
19.         <!-- custom MessageConverter define -->  
20.         <property name="messageConverter" ref="defaultMessageConverter" />  
21.     </bean>  

 8. MessageListenerContainer

 有幾個MessageListener既有幾個MessageListenerContainer

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.     <bean id="topicListenerContainerB" class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
7.         <property name="connectionFactory" ref="jmsConnectionFactory" />  
8.         <property name="destination" ref="TOPIC" />  
9.         <property name="messageListener" ref="topicListenerB" />  
10.     </bean>  
11.     <bean id="queueListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
12.         <property name="connectionFactory" ref="jmsConnectionFactory" />  
13.         <property name="destination" ref="QUEUE" />  
14.         <property name="messageListener" ref="queueListener" />  
15.     </bean>  

  Summary

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

1. 可以有一個或者多個訊息生產者向同一個destination傳送訊息.
2. queue型別的只能有一個訊息消費者.
3. topic型別的可以有多個訊息消費者.
4. 每個消費者對應一個MessageListener和一個MessageListenerContainer.

 這裡幫同事打個廣告：家裡種芒果，提前找批發零售渠道  可以以批發價個人預定

http://9community.com