ActiveMQ5.0實戰三:使用Spring發送,消費topic和queue消息

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

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

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

Xml代碼  

<!--  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代碼  

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

 

 3. JmsTemplate

Xml代碼  

<!--  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的實現見附件.

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

  5. MessageProducer

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

Xml代碼  

<!-- 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代碼  

<!--  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代碼  

<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代碼  

<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.