ActiveMQ5.0實(shí)戰(zhàn)一: 安裝配置ActiveMQ5.0

ActiveMQ5.0實(shí)戰(zhàn)二: 基本配置

簡介

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

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

?

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

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

Spring整合JMS

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

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. 消息消費(fèi)者.
7. MessageListener. 消息監(jiān)聽器
8. MessageListenerContainer. 消息監(jiān)聽容器

下面以實(shí)例的方式介紹上面8個(gè)部分.

1. ConnectionFactory

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

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

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

?在實(shí)例中我們使用了兩種destination

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

?3. JmsTemplate

    <!--  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實(shí)現(xiàn)的是org.springframework.jms.support.converter.MessageConverter接口, 提供消息的轉(zhuǎn)換功能. DefaultMessageConverter的實(shí)現(xiàn)見附件.

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

? 5. MessageProducer

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

    <!-- 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通道有兩個(gè)消息消費(fèi)者, QUEUE有一個(gè)消息消費(fèi)者

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

每一個(gè)消息消費(fèi)者都對(duì)應(yīng)一個(gè)MessageListener

    <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

?有幾個(gè)MessageListener既有幾個(gè)MessageListenerContainer

    <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配置文件的時(shí)候, 要把MessageProducer, MessageConsumer,MessageListener,MessageListenerContainer幾個(gè)地方弄清楚:

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

?

?

?

?

ActiveMQ5.0實(shí)戰(zhàn)三:使用Spring發(fā)送,消費(fèi)topic和queue消息