Terry.Li-彬

簡介

實戰一 , 實戰二介紹了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

Xml代碼

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

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

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

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實現的是org.springframework.jms.support.converter.MessageConverter接口, 提供消息的轉換功能. DefaultMessageConverter的實現見附件.

Xml代碼

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

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

  5. MessageProducer

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

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

每一個消息消費者都對應一個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. 可以有一個或者多個消息生產者向同一個destination發送消息.
2. queue類型的只能有一個消息消費者.
3. topic類型的可以有多個消息消費者.
4. 每個消費者對應一個MessageListener和一個MessageListenerContainer.