mq 使用Spring发送,消费topic和queue消息
2016-10-26 14:49
429 查看
简介
实战一 , 实战二介绍了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" />
[xml] view
plain copy
<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>
[xml] view
plain copy
<!-- 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
在实例中我们使用了两种destinationXml代码
<!-- ActiveMQ destinations -->
<!-- 使用topic方式-->
<amq:topic name="TOPIC" physicalName="JMS-TEST-TOPIC" />
<!-- 使用Queue方式-->
<amq:queue name="QUEUE" physicalName="JMS-TEST-QUEUE" />
[xml] view
plain copy
<!-- 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>
[xml] view
plain copy
<!-- 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" />
[xml] view
plain copy
<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>
[xml] view
plain copy
<!-- 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" />
[xml] view
plain copy
<!-- 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
每一个消息消费者都对应一个MessageListenerXml代码
<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>
[xml] view
plain copy
<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既有几个MessageListenerContainerXml代码
<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>
[xml] view
plain copy
<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.
相关文章推荐
- MQ系列3 使用Spring发送,消费topic和queue消息 activeMQ
- ActiveMQ5.0实战三:使用Spring发送,消费topic和queue消息
- ActiveMQ5.0实战三:使用Spring发送,消费topic和queue消息
- ActiveMQ5.0实战三:使用Spring发送,消费topic和queue消息
- ActiveMQ5.0实战三:使用Spring发送,消费topic和queue消息
- ActiveMQ5.0实战:使用Spring发送,消费topic和queue消息
- ActiveMQ5.0实战二:使用Spring发送,消费topic和queue消息
- spring-jms(activemq实现)使用queue发送消息简单例子
- springboot使用activemq同时接收queue和topic消息
- ActiveMQ 使用Queue或者Topic发送/接受消息
- spring整合activemq发送MQ消息[queue模式]实例
- spring整合activemq发送MQ消息[Topic模式]实例
- ActiveMQ (二) 使用Queue或者Topic发送/接受消息
- spring整合activemq发送MQ消息[Topic模式]实例,activemqmq
- 利用Spring与ActiveMQ整合发送、接收消息实例(Queue与Topic模式)
- ActiveMQ (二) 使用Queue或者Topic发送/接受消息
- rabbitmq学习9:使用spring-amqp发送消息及同步接受消息
- rabbitmq学习9:使用spring-amqp发送消息及同步接收消息
- RocketMQ3.2.2生产者发送消息自动创建Topic队列数无法超过4个
- Spring + ActiveMQ实现jms发送消息