`

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

    博客分类：
  • Java
SpringActiveMQBeanJMSXML 
阅读更多

简介

实战一 , 实战二介绍了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.

 

 

 

 

分享到：
| ActiveMQ5.0实战二: 基本配置
评论
1 楼 stevenjohn 2012-08-18  
楼主能上传一下源代码吗？
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    activeMq5.0

    在学习和使用ActiveMQ 5.0时，你可能需要了解如何创建和配置消息代理、如何编写生产者和消费者代码、如何设置安全策略、如何使用管理工具监控系统状态等。通过实践这些基本操作，你可以深入理解消息中间件的工作原理...

    ActiveMQ5.0 监视的JSP支持中文

    **标题：“ActiveMQ5.0 监视的JSP支持中文”** ActiveMQ是Apache软件基金会的一个开源项目，它是一个功能强大的消息中间件，广泛应用于分布式系统中的异步通信。在ActiveMQ 5.0版本中，对于中文支持的改进是一项...

    Apache ActiveMQ Queue Topic 详解

    ### Apache ActiveMQ Queue & Topic 详解 #### 一、特性及优势 Apache ActiveMQ 是一款高性能、功能丰富的消息中间件，具有以下显著特点： 1. **实现 JMS 1.1 规范**：支持 J2EE 1.4 及以上版本，这意味着它可以...

    activeMq 实战

    - 提供事务性上下文，使得一组消息的发送和接收操作可以作为单一的原子操作来执行。 ##### 1.4 目的地 (Destination) **目的地** 是客户端用来指定消息的发送目标和接收源的对象。根据 JMS 1.0.2 规范，有两种消息...

    ActiveMQ整合Spring(多消费者)

    同时，也需要配置目的地（Topic或Queue），可以使用`jms:topic`或`jms:queue`标签。 3. **消息生产者**： 生产者负责将消息放入消息队列。在Spring中，可以使用`JmsTemplate`的`send`方法来发送消息。需要指定目的...

    activemq与spring整合源代码

    5. 创建消息消费者：设置一个监听器容器，监听特定的队列或主题，当有新消息到达时，调用预先定义好的MessageListener进行处理。 四、实例分析 在提供的"mq"文件中，可能包含了一个示例项目，该项目展示了如何在...

    spring使用activeMQ实现消息发送

    5. **队列和主题的区别**：在ActiveMQ中，消息可以发送到队列（Queue）或主题（Topic）。队列遵循一对一模型，每个消息仅被一个消费者接收；主题遵循一对多模型，一个消息可以被多个订阅者接收。在上述示例中，我们...

    springboot2整合activemq的demo内含queue消息和topic消息

    - 在本地运行这个demo，首先确保安装并启动了ActiveMQ服务器，然后修改配置，运行Spring Boot应用，测试消息发送和接收功能。 - 部署时，可能需要考虑ActiveMQ集群、持久化存储、安全性等因素。 通过深入理解和...

    Spring平台整合消息队列ActiveMQ实现发布订阅、生产者消费者模型（适合新手或者开发人员了解学习ActiveMQ机制）

    本项目基于Spring这一平台，整合流行的开源消息队列中间件ActiveMQ,实现一个向ActiveMQ添加和读取消息的功能。并比较了两种模式：生产者-消费者模式和发布-订阅模式的区别。 包含的特性如下： 1.开启activeMQ，访问...

    Spring和ActiveMQ的整合实例源码

    5. **消息消费者（Consumer）**：Spring提供两种消费消息的方式：基于监听器的容器和基于回调的方法。前者通过实现`MessageListener`接口并在`@JmsListener`注解的回调方法中处理消息。后者则使用`JmsTemplate`的`...

    ActiveMQ-P2P文本消息+Spring和ActiveMQ的整合实例源码

    1. **Spring JMS**: Spring 提供了 JmsTemplate 类，它是发送和接收 JMS 消息的主要工具。通过配置，JmsTemplate 可以连接到 ActiveMQ 并进行消息操作。 2. **ConnectionFactory**: 这是 JMS 规范中的一个接口，...

    activemq_spring.rar_Spring和ActiveMQ_spring_消息中间件_消息发布订阅_消息订阅

    3. **配置Spring的JMS模板**：Spring的JmsTemplate是用于发送和接收消息的主要工具，我们需要在Spring配置文件中定义并配置它。 4. **创建消息生产者**：使用JmsTemplate的send方法可以发布消息到特定的队列或主题...

    activemq与spring整合发送jms消息入门实例

    3. **创建消息模板**：Spring的`JmsTemplate`是发送和接收JMS消息的核心工具。在配置文件中创建一个`JmsTemplate`的bean，并注入连接工厂。 ```xml &lt;bean id="jmsTemplate" class="org.springframework.jms.core....

    activemq 虚拟topic与路由功能

    在ActiveMQ消息中间件中，为了更好地实现消息的分发和管理，引入了虚拟Topic（Virtual Topic）的概念。本文将深入探讨虚拟Topic及路由功能，帮助读者理解如何利用这些特性来优化消息传递机制。 #### 一、虚拟Topic...

    SpringBoot+ActiveMq+MQTT实现消息的发送和接收

    4. 定义消息消费者：创建一个监听特定主题的MessageListener。当有消息到达时，这个监听器会被触发并处理消息。 5. 编写消息发送接口：为了将消息发送逻辑封装起来，可以创建一个服务类，提供发送消息的方法，这些...

    SpringActiveMQ.rar

    4. **消息消费者**：使用Spring的监听容器（如`DefaultMessageListenerContainer`），我们可以创建一个消息监听器来处理接收到的消息。监听器通常是一个实现了`MessageListener`接口的类，其中的`onMessage`方法会在...

    spring 整合 activemq 生产者和消费者 案例源码

    根据需求，配置相应的`Queue`或`Topic`，这将作为生产者发送消息和消费者接收消息的目标。 5. **配置MessageProducer和MessageConsumer**：在Spring配置中，定义`JmsTemplate`作为生产者，它可以发送消息到定义的...

    spring集成activemq演示queue和topic 持久化

    在本示例中，我们将深入探讨如何将Spring框架与ActiveMQ集成，以便实现消息队列（Queue）和主题（Topic）的功能，并确保消息的持久化。ActiveMQ是Apache软件基金会开发的一个开源消息中间件，它支持多种消息协议，如...

    spring 与ACTIVEMQ整合

    2. **创建消息消费者**：创建一个实现了`MessageListener`接口的Bean，重写`onMessage`方法，当接收到消息时执行相应的业务逻辑。 3. **注解方式消费**：使用`@JmsListener`注解，直接在方法上声明消息监听，简化...

Magicbox
Global site tag (gtag.js) - Google Analytics