`
dwj147258
  • 浏览: 194820 次
  • 性别: Icon_minigender_1
  • 来自: 深圳
社区版块
存档分类
最新评论

spring与activemq(搭建简单的消息发送与监听)

阅读更多

  一、JMS简介

          JMS的全称是Java Message Service,即Java消息服务。它主要用于在生产者和消费者之间进行消息传递,生产者负责产生消息,而消费者负责接收消息。把它应用到实际的业务需求中的话我们可以在特定的时候利用生产者生成一消息,并进行发送,对应的消费者在接收到对应的消息后去完成对应的业务逻辑。对于消息的传递有两种类型,一种是点对点的,即一个生产者和一个消费者一一对应;另一种是发布/订阅模式,即一个生产者产生消息并进行发送后,可以由多个消费者进行接收。

二、在Spring中配置JMS

      

1.2.1  activeMQ准备

       既然是使用的apacheactiveMQ作为JMS的实现,那么首先我们应该到apache官网上下载activeMQhttp://activemq.apache.org/download.html),进行解压后运行其bin目录下面的activemq.bat文件启动activeMQ

1.2.2配置ConnectionFactory

       ConnectionFactory是用于产生到JMS服务器的链接的,Spring为我们提供了多个ConnectionFactory,有SingleConnectionFactoryCachingConnectionFactorySingleConnectionFactory对于建立JMS服务器链接的请求会一直返回同一个链接,并且会忽略Connectionclose方法调用。CachingConnectionFactory继承了SingleConnectionFactory,所以它拥有SingleConnectionFactory的所有功能,同时它还新增了缓存功能,它可以缓存SessionMessageProducerMessageConsumer。这里我们使用SingleConnectionFactory来作为示例。

Xml代码  收藏代码
  1. <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"/>  

 

       这样就定义好产生JMS服务器链接的ConnectionFactory了吗?答案是非也。Spring提供的ConnectionFactory只是Spring用于管理ConnectionFactory的,真正产生到JMS服务器链接的ConnectionFactory还得是由JMS服务厂商提供,并且需要把它注入到Spring提供的ConnectionFactory中。我们这里使用的是ActiveMQ实现的JMS,所以在我们这里真正的可以产生Connection的就应该是由ActiveMQ提供的ConnectionFactory。所以定义一个ConnectionFactory的完整代码应该如下所示:

Xml代码  收藏代码
  1. <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->  
  2. <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
  3.     <property name="brokerURL" value="tcp://localhost:61616"/>  
  4. </bean>  
  5.   
  6. <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
  7. <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
  8.     <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
  9.     <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
  10. </bean>  

  

       ActiveMQ为我们提供了一个PooledConnectionFactory,通过往里面注入一个ActiveMQConnectionFactory可以用来将ConnectionSessionMessageProducer池化,这样可以大大的减少我们的资源消耗。当使用PooledConnectionFactory时,我们在定义一个ConnectionFactory时应该是如下定义:

 

Xml代码  收藏代码
  1. <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->  
  2. <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
  3.     <property name="brokerURL" value="tcp://localhost:61616"/>  
  4. </bean>  
  5.   
  6. <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">  
  7.     <property name="connectionFactory" ref="targetConnectionFactory"/>  
  8.     <property name="maxConnections" value="10"/>  
  9. </bean>  
  10.   
  11. <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
  12.     <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  
  13. </bean>  

1.2.3配置生产者

配置好ConnectionFactory之后我们就需要配置生产者。生产者负责产生消息并发送到JMS服务器,这通常对应的是我们的一个业务逻辑服务实现类。但是我们的服务实现类是怎么进行消息的发送的呢?这通常是利用Spring为我们提供的JmsTemplate类来实现的,所以配置生产者其实最核心的就是配置进行消息发送的JmsTemplate。对于消息发送者而言,它在发送消息的时候要知道自己该往哪里发,为此,我们在定义JmsTemplate的时候需要往里面注入一个Spring提供的ConnectionFactory对象。

Xml代码  收藏代码
  1. <!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->  
  2. <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
  3.     <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
  4.     <property name="connectionFactory" ref="connectionFactory"/>  
  5. </bean>  

 

       在真正利用JmsTemplate进行消息发送的时候,我们需要知道消息发送的目的地,即destination。在Jms中有一个用来表示目的地的Destination接口,它里面没有任何方法定义,只是用来做一个标识而已。当我们在使用JmsTemplate进行消息发送时没有指定destination的时候将使用默认的Destination。默认Destination可以通过在定义jmsTemplate bean对象时通过属性defaultDestinationdefaultDestinationName来进行注入,defaultDestinationName对应的就是一个普通字符串。在ActiveMQ中实现了两种类型的Destination,一个是点对点的ActiveMQQueue,另一个就是支持订阅/发布模式的ActiveMQTopic。在定义这两种类型的Destination时我们都可以通过一个name属性来进行构造,如:

 

Xml代码  收藏代码
  1. <!--这个是队列目的地,点对点的-->  
  2. <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
  3.     <constructor-arg>  
  4.         <value>queue</value>  
  5.     </constructor-arg>  
  6. </bean>  
  7. <!--这个是主题目的地,一对多的-->  
  8. <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">  
  9.     <constructor-arg value="topic"/>  
  10. </bean>  

       假设我们定义了一个ProducerService,里面有一个向Destination发送纯文本消息的方法sendMessage,那么我们的代码就大概是这个样子:

 

Java代码  收藏代码
  1. package com.tiantian.springintejms.service.impl;  
  2.    
  3. import javax.annotation.Resource;  
  4. import javax.jms.Destination;  
  5. import javax.jms.JMSException;  
  6. import javax.jms.Message;  
  7. import javax.jms.Session;  
  8.    
  9. import org.springframework.jms.core.JmsTemplate;  
  10. import org.springframework.jms.core.MessageCreator;  
  11. import org.springframework.stereotype.Component;  
  12.    
  13. import com.tiantian.springintejms.service.ProducerService;  
  14.    
  15. @Component  
  16. public class ProducerServiceImpl implements ProducerService {  
  17.    
  18.     private JmsTemplate jmsTemplate;  
  19.       
  20.     public void sendMessage(Destination destination, final String message) {  
  21.         System.out.println("---------------生产者发送消息-----------------");  
  22.         System.out.println("---------------生产者发了一个消息:" + message);  
  23.         jmsTemplate.send(destination, new MessageCreator() {  
  24.             public Message createMessage(Session session) throws JMSException {  
  25.                 return session.createTextMessage(message);  
  26.             }  
  27.         });  
  28.     }   
  29.   
  30.     public JmsTemplate getJmsTemplate() {  
  31.         returnjmsTemplate;  
  32.     }   
  33.   
  34.     @Resource  
  35.     public void setJmsTemplate(JmsTemplate jmsTemplate) {  
  36.         this.jmsTemplate = jmsTemplate;  
  37.     }  
  38.    
  39. }  

       我们可以看到在sendMessage方法体里面我们是通过jmsTemplate来发送消息到对应的Destination的。到此,我们生成一个简单的文本消息并把它发送到指定目的地Destination的生产者就配置好了。

1.2.4配置消费者

生产者往指定目的地Destination发送消息后,接下来就是消费者对指定目的地的消息进行消费了。那么消费者是如何知道有生产者发送消息到指定目的地Destination了呢?这是通过Spring为我们封装的消息监听容器MessageListenerContainer实现的,它负责接收信息,并把接收到的信息分发给真正的MessageListener进行处理。每个消费者对应每个目的地都需要有对应的MessageListenerContainer。对于消息监听容器而言,除了要知道监听哪个目的地之外,还需要知道到哪里去监听,也就是说它还需要知道去监听哪个JMS服务器,这是通过在配置MessageConnectionFactory的时候往里面注入一个ConnectionFactory来实现的。所以我们在配置一个MessageListenerContainer的时候有三个属性必须指定,一个是表示从哪里监听的ConnectionFactory;一个是表示监听什么的Destination;一个是接收到消息以后进行消息处理的MessageListenerSpring一共为我们提供了两种类型的MessageListenerContainerSimpleMessageListenerContainerDefaultMessageListenerContainer

SimpleMessageListenerContainer会在一开始的时候就创建一个会话session和消费者Consumer,并且会使用标准的JMS MessageConsumer.setMessageListener()方法注册监听器让JMS提供者调用监听器的回调函数。它不会动态的适应运行时需要和参与外部的事务管理。兼容性方面,它非常接近于独立的JMS规范,但一般不兼容Java EEJMS限制。

大多数情况下我们还是使用的DefaultMessageListenerContainer,跟SimpleMessageListenerContainer相比,DefaultMessageListenerContainer会动态的适应运行时需要,并且能够参与外部的事务管理。它很好的平衡了对JMS提供者要求低、先进功能如事务参与和兼容Java EE环境。

定义处理消息的MessageListener

       要定义处理消息的MessageListener我们只需要实现JMS规范中的MessageListener接口就可以了。MessageListener接口中只有一个方法onMessage方法,当接收到消息的时候会自动调用该方法。

 

Java代码  收藏代码
  1. package com.tiantian.springintejms.listener;  
  2.    
  3. import javax.jms.JMSException;  
  4. import javax.jms.Message;  
  5. import javax.jms.MessageListener;  
  6. import javax.jms.TextMessage;  
  7.    
  8. public class ConsumerMessageListener implements MessageListener {  
  9.    
  10.     public void onMessage(Message message) {  
  11.         //这里我们知道生产者发送的就是一个纯文本消息,所以这里可以直接进行强制转换  
  12.         TextMessage textMsg = (TextMessage) message;  
  13.         System.out.println("接收到一个纯文本消息。");  
  14.         try {  
  15.             System.out.println("消息内容是:" + textMsg.getText());  
  16.         } catch (JMSException e) {  
  17.             e.printStackTrace();  
  18.         }  
  19.     }  
  20.    
  21. }  

 

       有了MessageListener之后我们就可以在Spring的配置文件中配置一个消息监听容器了。

Xml代码  收藏代码
  1. <!--这个是队列目的地-->  
  2. <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
  3.     <constructor-arg>  
  4.         <value>queue</value>  
  5.     </constructor-arg>  
  6. </bean>  
  7. <!-- 消息监听器 -->  
  8. <bean id="consumerMessageListener" class="com.tiantian.springintejms.listener.ConsumerMessageListener"/>      
  9.   
  10. <!-- 消息监听容器 -->  
  11. <bean id="jmsContainer"        class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
  12.     <property name="connectionFactory" ref="connectionFactory" />  
  13.     <property name="destination" ref="queueDestination" />  
  14.     <property name="messageListener" ref="consumerMessageListener" />  
  15. </bean>  

  

       我们可以看到我们定义了一个名叫queueActiveMQQueue目的地,我们的监听器就是监听了发送到这个目的地的消息。

       至此我们的生成者和消费者都配置完成了,这也就意味着我们的整合已经完成了。这个时候完整的Spring的配置文件应该是这样的:

<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"  
    xmlns:jms="http://www.springframework.org/schema/jms"  
    xsi:schemaLocation="http://www.springframework.org/schema/beans  
     http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
     http://www.springframework.org/schema/context  
     http://www.springframework.org/schema/context/spring-context-3.0.xsd  
    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd  
    http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd">  
   
    <context:component-scan base-package="com.tiantian" />  
   
    <!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->  
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
        <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
        <property name="connectionFactory" ref="connectionFactory"/>  
    </bean>  
      
    <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->  
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
        <property name="brokerURL" value="tcp://localhost:61616"/>  
    </bean>  
      
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
    </bean>  
      
    <!--这个是队列目的地-->  
    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
        <constructor-arg>  
            <value>queue</value>  
        </constructor-arg>  
    </bean>  
    <!-- 消息监听器 -->  
    <bean id="consumerMessageListener" class="com.tiantian.springintejms.listener.ConsumerMessageListener"/>  
    <!-- 消息监听容器 -->  
    <bean id="jmsContainer"  
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
        <property name="connectionFactory" ref="connectionFactory" />  
        <property name="destination" ref="queueDestination" />  
        <property name="messageListener" ref="consumerMessageListener" />  
    </bean>  
</beans>  
 三、另一种配置,
<!--   方式一  start -->
    <!-- ActiveMQ 连接工厂 -->
    <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
    <!-- 如果连接网络:tcp://ip:61616;未连接网络:tcp://localhost:61616 以及用户名,密码-->
  	 <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin"/>
	
    <!-- Spring Caching连接工厂 -->
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
        <property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
        <!-- 同上,同理 -->
        <!-- <constructor-arg ref="amqConnectionFactory" /> -->
        <!-- Session缓存数量 -->
        <property name="sessionCacheSize" value="100" />
    </bean>
    
    <!-- Spring JmsTemplate 的消息生产者 start-->

    <!-- 定义JmsTemplate的Queue类型 -->
    <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
        <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
        <constructor-arg ref="connectionFactory" />
        <!-- 非pub/sub模型(发布/订阅),即队列模式 -->
        <property name="pubSubDomain" value="false" />
    </bean>

    <!-- 定义JmsTemplate的Topic类型 -->
    <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
         <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
        <constructor-arg ref="connectionFactory" />
        <!-- pub/sub模型(发布/订阅) -->
        <property name="pubSubDomain" value="true" />
    </bean>
    
    <!-- 消息消费者 start-->
	<bean id="queueReceiver1" class="main.java.test.jms.SpringActivemq.mq.consumer.queue.QueueReceiver1"/>
	<bean id="queueReceiver2" class="main.java.test.jms.SpringActivemq.mq.consumer.queue.QueueReceiver2"/>
	<bean id="topicReceiver1" class="main.java.test.jms.SpringActivemq.mq.consumer.topic.TopicReceiver1"/>
	<bean id="topicReceiver2" class="main.java.test.jms.SpringActivemq.mq.consumer.topic.TopicReceiver2"/>
    <!-- 定义Queue监听器 -->
    <jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
        <jms:listener destination="test.queue" ref="queueReceiver1"/>
        <jms:listener destination="test.queue" ref="queueReceiver2"/>
    </jms:listener-container>

    <!-- 定义Topic监听器 -->
    <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
        <jms:listener destination="hello" ref="topicReceiver1" method="hand"/>
        <jms:listener destination="topic" ref="topicReceiver2"/>
    </jms:listener-container>
    <!-- 消息消费者 end -->
    <!-- 方式一  end -->
 发送topic消息还是queue消息是通过配置jmstemple的时候指定pubSubDomain属性,为true则是发送topic消息。在这里 ,详见http://blog.csdn.net/jiuqiyuliang/article/details/48758203这样就可以通过activemq来发送消息了, 

 

分享到:
评论

相关推荐

    Spring 和 activemq 搭建JMS开发系统示例

    **Spring与ActiveMQ搭建JMS开发系统示例详解** 在Java世界中,消息队列(JMS,Java Message Service)是一种标准,它定义了API来创建、发送、接收和读取消息,允许应用程序进行异步通信。Spring框架是Java开发中的...

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

    3. 创建消息生产者:创建一个类,使用Spring的JMSTemplate或者MQTT客户端API来发送消息。JMSTemplate提供了一种简单的方式来发送JMS消息,而MQTT客户端API则用于发送MQTT消息。 4. 定义消息消费者:创建一个监听...

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

    Spring整合ActiveMQ是Java消息服务(JMS)在Spring框架中的应用,用于实现生产者与消费者的解耦。在这个案例中,我们将深入探讨如何配置和使用这两个组件,以便于理解它们的工作原理。 首先,ActiveMQ是Apache软件...

    SpringBoot整合ActiveMQ+websocket.docx

    在`QueueProduce`类中,使用了`JmsMessagingTemplate`来发送消息到队列,`convertAndSend()`方法将数据转换为消息并发送出去。这里有两个方法,一个是即时发送`produceMsg()`,另一个是定时发送`produceMsgScheduled...

    消息队列介绍和SpringBoot2.x整合RockketMQ、ActiveMQ

    生产者使用`JmsTemplate`的`send()`方法发送消息到特定的Queue,而消费者通过`@JmsListener`监听这个Queue,当有新消息到达时,会自动调用指定的方法进行处理。 **总结** 本文介绍了消息队列的基本概念和作用,...

    Spring+JMS+ActiveMQ+Tomcat实现消息服务_服务器应用

    用户可以通过调用 `SendMessageController` 类中的 `send` 方法来发送消息,而这些消息会被 `ReceiveMessageListener` 类接收并处理。这种架构不仅简洁而且高效,非常适合构建实时消息传递系统。

    java消息中间件教程-activemq

    - Spring JMS是Spring框架中用于与JMS服务交互的模块,提供了高级抽象层来简化消息处理。 - **4-6 Spring JMS演示** - 创建一个名为`jms-spring`的Maven项目,并添加必要的依赖。 - 编写消息生产者服务类`...

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

    - 可能包含`@JmsListener`注解用于定义消息监听器,以及`JmsTemplate`用于发送消息。 8. **开发实践**: - 开发者在实际项目中,可以根据需求选择Queue或Topic,编写相应的消息处理器,并调整ActiveMQ配置以适应...

    apache-activemq-5.17.5

    3. 创建消息生产者:通过Spring Boot的自动配置,我们可以轻松创建一个消息生产者,发送消息到ActiveMQ队列或主题: ```java @Autowired private JmsTemplate jmsTemplate; public void sendMessage(String message...

    Springboot+ActiveMq

    创建一个服务类,使用`JmsTemplate`发送消息到指定的队列。例如: ```java @Autowired private JmsTemplate jmsTemplate; public void sendMessage(String queueName, String message) { jmsTemplate....

    ActiveMQ整合spring、SpringBoot完整源码

    Spring对JMS提供了一套抽象层,简化了与消息中间件的交互。在配置中,我们需要在Spring的XML配置文件或Java配置类中定义一个ConnectionFactory,这是连接到ActiveMQ服务器的桥梁。接着,创建Destination(主题或队列...

    Apache+ActiveMQ教程

    本教程将深入探讨如何利用Apache ActiveMQ搭建、配置和集成到Spring框架中,以实现高效的数据通信。 首先,我们需要了解JMS的基本概念。Java消息服务(JMS)是Java平台上的一个标准接口,它定义了应用程序如何创建...

    springboot-activemq-demo.zip

    在SpringBoot应用中创建一个消息生产者,使用`JmsTemplate`发送消息到ActiveMQ的队列或主题。例如: ```java import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms....

    boot activemq

    3. **创建MessageProducer**:在Spring Boot应用中,我们通常会创建一个`@Service`层的类,其中包含方法来发送消息。这些方法可以使用`JmsTemplate`或`JmsMessagingTemplate`来创建`MessageProducer`,并通过它将...

    ActiveMQ入门

    - **发送消息**:在商品上架成功后,调用相应的消息发送接口,将商品ID发送到指定的队列中。 - **接收消息**:索引服务监听队列,一旦接收到新的消息,就立即处理并更新索引。 #### 四、结论 ActiveMQ作为一种高...

    springboot集成activeMQ

    在Spring Boot中,我们可以使用JmsTemplate发送消息到队列,然后通过@JmsListener监听队列并消费消息: ```java @Service public class ProducerService { @Autowired private JmsTemplate jmsTemplate; ...

    springboot activemq

    这里,我们创建了一个JmsTemplate实例用于发送消息,并配置了一个监听器容器来处理接收到的消息。`MyMessageListener`是你自定义的消息监听器类,需要实现`javax.jms.MessageListener`接口。 对于topic的配置,只需...

    Spring整合JMS.doc

    Spring整合JMS与ActiveMQ深度解析 一、JMS概览与原理 Java Message Service (JMS) 是Java平台中的消息服务应用接口,为应用程序之间的通信提供消息传递机制。JMS支持两种通信模型:点对点(Point-to-Point, PTP)...

    SpringBoot整合ActiveMQ完整源码分享给需要的同学

    Spring对JMS提供了一套抽象层,简化了与消息中间件的交互。在配置中,我们需要在Spring的XML配置文件或Java配置类中定义一个ConnectionFactory,这是连接到ActiveMQ服务器的桥梁。接着,创建Destination(主题或队列...

Global site tag (gtag.js) - Google Analytics