2. it
  3. ActiveMQ5.0实战三:使用Spring发送,消费topic和queue消息

ActiveMQ5.0实战三:使用Spring发送,消费topic和queue消息

mac2022-06-30  231

实战一 , 实战二 介绍了ActiveMQ的基本概念和配置方式.

本篇将通过一个实例介绍使用spring发送,消费topic, queue类型消息的方法. 不懂topic和queue的google 之.

 

如图示, TOPIC和QUEUE分别代表一个topic和一个queue消息通道.

TopicMessageProducer向topic发送消息, TopicConsumerA和TopicConsumerB则从topic消费消息.QueueMessageProducer向Queue发送消息, QueueConsumer从Queue中消费消息

Spring整合JMS

就像对orm, web的支持一样, spring同样支持jms, 为整合jms到已有的项目提供了很多便利的方法. 本篇主要讲实战, 是所以先从配置开始, spring配置jms基本上需要8个部分.

ConnectionFactory. 和jms服务器的连接, 可以是外部的jms server, 也可以使用embedded ActiveMQ Broker.Destination. 有topic和queue两种方式.JmsTemplate. spring提供的jms模板.MessageConverter. 消息转换器.MessageProducer. 消息生产者.MessageConsumer. 消息消费者.MessageListener. 消息监听器MessageListenerContainer. 消息监听容器

下面以实例的方式介绍上面8个部分.

1. ConnectionFactory

Xml代码   <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的实现见附件.

Xml代码   <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几个地方弄清楚:

可以有一个或者多个消息生产者向同一个destination发送消息.queue类型的只能有一个消息消费者.topic类型的可以有多个消息消费者.每个消费者对应一个MessageListener和一个MessageListenerContainer.

转载于:https://www.cnblogs.com/kms1989/p/7525249.html

相关资源：JAVA上百实例源码以及开源项目

专利

最新回复(0)