JMS-ActiveMq与Spring整合

矫情吗；* 2022-04-10 12:37 207阅读 0赞

JMS-ActiveMq与Spring整合:此处代码直接就可以用

工具:

IDEA,

jdk1.8,

Activemq5.14

Tomcat8

Gradle

项目的配置  
首先得是Gradle配置(此处采用Gradle依赖管理):

"org.springframework:spring-jms:4.2.4.RELEASE",  
"org.apache.activemq:activemq-all:5.14.2",  
"org.springframework:spring-web:4.1.4.RELEASE",  
"org.springframework:spring-core:4.1.4.RELEASE",  
"org.springframework:spring-webmvc:4.1.4.RELEASE",  
"org.springframework:spring-tx:4.1.4.RELEASE",

配置ConnectionFactory  
connectionFactory是Spring用于创建到JMS服务器链接的，Spring提供了多种connectionFactory，我们介绍两个SingleConnectionFactory和CachingConnectionFactory。

SingleConnectionFactory：对于建立JMS服务器链接的请求会一直返回同一个链接  
，并且会忽略Connection的close方法调用。

CachingConnectionFactory：继承了SingleConnectionFactory，所以它拥有SingleConnectionFactory的所有功能，同时它还新增了缓存功能，它可以缓存Session、MessageProducer和MessageConsumer。我们使用CachingConnectionFactory来作为示例。

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

<amq:connectionFactory id="amqConnectionFactory"  
        brokerURL="tcp://192.168.3.3:61616" userName="admin" password="admin"  />  
   
      
        
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  
            
        <property name="targetConnectionFactory" ref="amqConnectionFactory"></property>  
          
          
          
        <property name="sessionCacheSize" value="100" />  
    </bean>

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

在利用JmsTemplate进行消息发送的时候，我们需要知道发送哪种消息类型：一个是点对点的ActiveMQQueue，另一个就是支持订阅/发布模式的ActiveMQTopic。如下所示：

<bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">  
            
        <constructor-arg ref="connectionFactory" />  
          
        <property name="pubSubDomain" value="false" />  
    </bean>  
   
      
    <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">  
             
        <constructor-arg ref="connectionFactory" />  
          
        <property name="pubSubDomain" value="true" />  
    </bean>

生产者如何指定目的地和发送消息？大家看源码即可，就不再这提供了。

配置消费者  
生产者往指定目的地Destination发送消息后，接下来就是消费者对指定目的地的消息进行消费了。那么消费者是如何知道有生产者发送消息到指定目的地Destination了呢？每个消费者对应每个目的地都需要有对应的MessageListenerContainer。对于消息监听容器而言，除了要知道监听哪个目的地之外，还需要知道到哪里去监听，也就是说它还需要知道去监听哪个JMS服务器，通过配置MessageListenerContainer的时候往里面注入一个ConnectionFactory来实现的。所以我们在配置一个MessageListenerContainer的时候有三个属性必须指定：一个是表示从哪里监听的ConnectionFactory；一个是表示监听什么的Destination；一个是接收到消息以后进行消息处理的MessageListener。

<bean id="queueReceiver1" class="com.lei.activemq.springmq.comsumer.queue.QueueReceiver1"/>  
    <bean id="queueReceiver2" class="com.lei.activemq.springmq.comsumer.queue.QueueReceiver2"/>  
      
    <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>  
   
    <bean id="topicReceiver1" class="com.lei.activemq.springmq.comsumer.topic.TopicReceiver1"/>  
    <bean id="topicReceiver2" class="com.lei.activemq.springmq.comsumer.topic.TopicReceiver2"/>  
      
    <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">  
        <jms:listener destination="test.topic" ref="topicReceiver1"/>  
        <jms:listener destination="test.topic" ref="topicReceiver2"/>  
    </jms:listener-container>

此时，Spring和JMS，ActiveMQ整合的spring-activemq.xml已经完成，下面展示所有的xml

<?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:amq="http://activemq.apache.org/schema/core"  
       xmlns:jms="http://www.springframework.org/schema/jms"  
       xsi:schemaLocation="http://www.springframework.org/schema/beans  
        http://www.springframework.org/schema/beans/spring-beans-4.0.xsd  
        http://www.springframework.org/schema/jms  
        http://www.springframework.org/schema/jms/spring-jms-4.0.xsd  
        http://activemq.apache.org/schema/core  
        http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd">  
      
      
      
    <amq:connectionFactory id="amqConnectionFactory"  
                           brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin"  />  
      
      
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  
          
        <property name="targetConnectionFactory" ref="amqConnectionFactory"></property>  
          
          
          
        <property name="sessionCacheSize" value="100" />  
    </bean>  
      
   
      
    <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">  
          
        <constructor-arg ref="connectionFactory" />  
          
        <property name="pubSubDomain" value="false" />  
    </bean>  
   
      
    <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">  
          
        <constructor-arg ref="connectionFactory" />  
          
        <property name="pubSubDomain" value="true" />  
    </bean>  
   
      
      
    <bean id="queueReceiver1" class="com.lei.activemq.springmq.comsumer.queue.QueueReceiver1"/>  
    <bean id="queueReceiver2" class="com.lei.activemq.springmq.comsumer.queue.QueueReceiver2"/>  
      
    <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>  
   
    <bean id="topicReceiver1" class="com.lei.activemq.springmq.comsumer.topic.TopicReceiver1"/>  
    <bean id="topicReceiver2" class="com.lei.activemq.springmq.comsumer.topic.TopicReceiver2"/>  
      
    <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">  
        <jms:listener destination="test.topic" ref="topicReceiver1"/>  
        <jms:listener destination="test.topic" ref="topicReceiver2"/>  
    </jms:listener-container>  
   
      
   
   
</beans>

spring-core.xml配置:  
<?xml version="1.0" encoding="UTF-8"?>  
<beans xmlns="http://www.springframework.org/schema/beans"  
       xmlns:context="http://www.springframework.org/schema/context"  
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
       xsi:schemaLocation="http://www.springframework.org/schema/beans  
        http://www.springframework.org/schema/beans/spring-beans-4.0.xsd  
        http://www.springframework.org/schema/context  
        http://www.springframework.org/schema/context/spring-context-4.0.xsd">  
    <context:annotation-config />  
   <import resource="spring-activemq.xml"/>  
   
   
</beans>  
spring-mvc.xml配置:  
<?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:c="http://www.springframework.org/schema/c"  
       xmlns:context="http://www.springframework.org/schema/context"  
       xmlns:mvc="http://www.springframework.org/schema/mvc"  
       xmlns:p="http://www.springframework.org/schema/p"  
       xsi:schemaLocation="http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd  
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd  
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd">  
    <mvc:annotation-driven />  
    <mvc:default-servlet-handler/>  
      
      
    <context:component-scan base-package="com.lei.activemq"/>  
   
   
      
    <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver">  
        <property name="prefix" value="/WEB-INF/views/" />  
        <property name="suffix" value=".jsp" />  
         
        <property name="order" value="1" />  
    </bean>  
   
</beans>

web.xml配置:  
<?xml version="1.0" encoding="UTF-8"?>  
<web-app xmlns="http://java.sun.com/xml/ns/javaee"  
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee  
          http://java.sun.com/xml/ns/javaee/web-app\_3\_0.xsd"  
         version="3.0">  
   
      
    <filter>  
        <filter-name>CharacterEncodingFilter</filter-name>  
        <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>  
        <init-param>  
            <param-name>encoding</param-name>  
            <param-value>utf-8</param-value>  
        </init-param>  
    </filter>  
    <filter-mapping>  
        <filter-name>CharacterEncodingFilter</filter-name>  
        <url-pattern>/\*</url-pattern>  
    </filter-mapping>  
      
    <servlet>  
        <servlet-name>dispatcher</servlet-name>  
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>  
        <init-param>  
            <param-name>contextConfigLocation</param-name>  
            <param-value>classpath\*:spring-mvc.xml</param-value>  
        </init-param>  
        <load-on-startup>1</load-on-startup>  
    </servlet>  
   
    <servlet-mapping>  
        <servlet-name>dispatcher</servlet-name>  
        <url-pattern>/</url-pattern>  
    </servlet-mapping>  
   
      
    <listener>  
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>  
    </listener>  
   
    <session-config>  
        <session-timeout>10</session-timeout>  
    </session-config>  
   
    <context-param>  
        <param-name>contextConfigLocation</param-name>  
        <param-value>classpath\*:spring-core.xml</param-value>  
    </context-param>  
   
      
    <welcome-file-list>  
        <welcome-file>index.jsp</welcome-file>  
    </welcome-file-list>  
</web-app>

生产者代码QueueSender.java:

package com.lei.activemq.springmq.producer.queue;  
   
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.beans.factory.annotation.Qualifier;  
import org.springframework.jms.core.JmsTemplate;  
import org.springframework.jms.core.MessageCreator;  
import org.springframework.stereotype.Component;  
   
import javax.jms.JMSException;  
import javax.jms.Message;  
import javax.jms.Session;  
   
/\*\*  
 \* Created by leitao on 2017/3/2.  
 \*/  
@Component("queueSender")  
public class QueueSender \{  
   
    @Autowired  
    @Qualifier("jmsQueueTemplate")  
    private JmsTemplate jmsTemplate;//通过@Qualifier修饰符来注入对应的bean  
   
    /\*\*  
     \* 发送一条消息到指定的队列（目标）  
     \* @param queueName 队列名称  
     \* @param message 消息内容  
     \*/  
    public void send(String queueName,final String message)\{  
        jmsTemplate.send(queueName, new MessageCreator() \{  
            @Override  
            public Message createMessage(Session session) throws JMSException \{  
                return session.createTextMessage(message);  
            \}  
        \});  
    \}  
   
\}

生产者TopicSender.java:  
package com.lei.activemq.springmq.producer.topic;  
   
import org.springframework.beans.factory.annotation.Autowired;  
import org.springframework.beans.factory.annotation.Qualifier;  
import org.springframework.jms.core.JmsTemplate;  
import org.springframework.jms.core.MessageCreator;  
import org.springframework.stereotype.Component;  
   
import javax.jms.JMSException;  
import javax.jms.Message;  
import javax.jms.Session;  
   
/\*\*  
 \* Created by leitao on 2017/3/2.  
 \*/  
@Component("topicSender")  
public class TopicSender \{  
   
    @Autowired  
    @Qualifier("jmsTopicTemplate")  
    private JmsTemplate jmsTemplate;  
   
    /\*\*  
     \* 发送一条消息到指定的队列（目标）  
     \* @param topicName 队列名称  
     \* @param message 消息内容  
     \*/  
    public void send(String topicName,final String message)\{  
        jmsTemplate.send(topicName, new MessageCreator() \{  
            @Override  
            public Message createMessage(Session session) throws JMSException \{  
                return session.createTextMessage(message);  
            \}  
        \});  
    \}  
   
\}  
消费者代码如下:  
package com.lei.activemq.springmq.comsumer.queue;  
   
import org.springframework.stereotype.Component;  
   
import javax.jms.JMSException;  
import javax.jms.Message;  
import javax.jms.MessageListener;  
import javax.jms.TextMessage;  
   
/\*\*  
 \* Created by leitao on 2017/3/2.  
 \*/  
@Component  
public class QueueReceiver1 implements MessageListener \{  
   
    @Override  
    public void onMessage(Message message) \{  
        try \{  
            System.out.println("QueueReceiver1接收到消息:"+((TextMessage)message).getText());  
        \} catch (JMSException e) \{  
            e.printStackTrace();  
        \}  
    \}  
\}  
package com.lei.activemq.springmq.comsumer.queue;  
   
import org.springframework.stereotype.Component;  
   
import javax.jms.JMSException;  
import javax.jms.Message;  
import javax.jms.MessageListener;  
import javax.jms.TextMessage;  
   
/\*\*  
 \* Created by leitao on 2017/3/2.  
 \*/  
@Component  
public class QueueReceiver2 implements MessageListener\{  
    @Override  
    public void onMessage(Message message) \{  
        try \{  
            System.out.println("QueueReceiver2接收到消息:"+((TextMessage)message).getText());  
        \} catch (JMSException e) \{  
            e.printStackTrace();  
        \}  
    \}  
\}

package com.lei.activemq.springmq.comsumer.topic;  
   
import org.springframework.stereotype.Component;  
   
import javax.jms.JMSException;  
import javax.jms.Message;  
import javax.jms.MessageListener;  
import javax.jms.TextMessage;  
   
/\*\*  
 \* Created by leitao on 2017/3/2.  
 \*/  
@Component  
public class TopicReceiver1 implements MessageListener \{  
    @Override  
    public void onMessage(Message message) \{  
        try \{  
            System.out.println("TopicReceiver1接收到消息:"+((TextMessage)message).getText());  
        \} catch (JMSException e) \{  
            e.printStackTrace();  
        \}  
    \}  
\}  
package com.lei.activemq.springmq.comsumer.topic;  
   
import org.springframework.stereotype.Component;  
   
import javax.jms.JMSException;  
import javax.jms.Message;  
import javax.jms.MessageListener;  
import javax.jms.TextMessage;  
   
/\*\*  
 \* Created by leitao on 2017/3/2.  
 \*/  
@Component  
public class TopicReceiver2 implements MessageListener\{  
    @Override  
    public void onMessage(Message message) \{  
        try \{  
            System.out.println("TopicReceiver2接收到消息:"+((TextMessage)message).getText());  
        \} catch (JMSException e) \{  
            e.printStackTrace();  
        \}  
    \}  
\}

测试Controller:  
package com.lei.activemq.springmq;  
   
import com.lei.activemq.springmq.producer.queue.QueueSender;  
import com.lei.activemq.springmq.producer.topic.TopicSender;  
import org.springframework.stereotype.Controller;  
import org.springframework.web.bind.annotation.RequestMapping;  
import org.springframework.web.bind.annotation.RequestParam;  
import org.springframework.web.bind.annotation.ResponseBody;  
   
import javax.annotation.Resource;  
   
/\*\*  
 \* Created by leitao on 2017/3/2.  
 \*/  
@Controller  
@RequestMapping(value = "activemq")  
public class ActivemqController \{  
   
    @Resource  
    QueueSender queueSender;  
    @Resource  
    TopicSender topicSender;  
   
    /\*\*  
     \* 发送消息到队列  
     \* Queue队列：仅有一个订阅者会收到消息，消息一旦被处理就不会存在队列中  
     \* @param message  
     \* @return String  
     \*/  
    @ResponseBody  
    @RequestMapping(value = "/queueSender")  
    public String queueSender(@RequestParam("message")String message)\{  
        String opt="";  
        try \{  
            queueSender.send("test.queue", message);  
            opt = "suc";  
        \} catch (Exception e) \{  
            opt = e.getCause().toString();  
        \}  
        return opt;  
    \}  
   
    /\*\*  
     \* 发送消息到主题  
     \* Topic主题 ：放入一个消息，所有订阅者都会收到  
     \* 这个是主题目的地是一对多的  
     \* @param message  
     \* @return String  
     \*/  
    @ResponseBody  
    @RequestMapping(value = "/topicSender")  
    public String topicSender(@RequestParam("message")String message)\{  
        String opt = "";  
        try \{  
            topicSender.send("test.topic", message);  
            opt = "suc";  
        \} catch (Exception e) \{  
            opt = e.getCause().toString();  
        \}  
        return opt;  
    \}  
\}

源码下载:http://download.csdn.net/detail/admin1973/9768550

参考资料:

http://blog.csdn.net/jiuqiyuliang/article/details/48758203  
\---------------------   
作者：涛濤   
来源：CSDN   
原文：https://blog.csdn.net/admin1973/article/details/59540386   
版权声明：本文为博主原创文章，转载请附上博文链接！