activemq与spring整合

消息队列activemq实现了jms标准，而且还可以与spring进行无缝整合，到现在activemq仍然活跃在企业的项目中。在说明activemq与spring整合前，首先来了解一下基本概念。源代码链接

1、JMS

JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API。Activemq消息队列组件实现了JMS操作规范。

2、JMS消息通信方式

通常有两种方式：点对点（Point-to-Point）和发布与订阅（Publish/Subscribe）。点对点的方式类似与日常生活正的两个手机用户之间的互发短信，只能是一个发送另一个接收；发布与订阅类似于我们关注的公众号，一个公众号可以被多个用户关注和订阅，每当公众号有更新后，我们打开app都会接收到相关推送的消息。

3、JMS操作流程

按照JMS的规范，我们首先需要获得一个JMS connection factory.，通过这个connection factory来创建connection.在这个基础之上我们再创建session, destination, producer和consumer。因此主要的几个步骤如下：

（1）. 获得JMS connection factory. 通过我们提供特定环境的连接信息来构造factory。

（2）. 利用factory构造JMS connection

（3）. 启动connection

（4）. 通过connection创建JMS session.

（5）. 指定JMS destination.

（6）. 创建JMS producer或者创建JMS message并提供destination.

（7）. 创建JMS consumer或注册JMS message listener.

（8）. 发送和接收JMS message.

（9）. 关闭所有JMS资源，包括connection, session, producer, consumer等。

4、下载安装activeMQ

Activemq下载地址：http://activemq.apache.org/download-archives.html

可下载的版本分为window版和Linux版，本次下载的为linux版本的activemq：apache-activemq-5.9.1-bin.tar.gz。

（1）、将active mq组件上传到Linux下，而后解压缩：

tar xzvf /srv/ftp/apache-activemq-5.12.0-bin.tar.gz -C /usr/local

（2）、配置一个新的用户：activemq

· 修改文件：vim /usr/local/apache-activemq-5.12.0/conf/jetty-realm.properties；

activemq : java,user

activemq与spring整合

（3）、启动ActiveMQ服务：

/usr/local/apache-activemq-5.12.0/bin/activemq start

（4）、打开页面浏览器：输入http://IP:8161，看到以下界面说明activemq安装成功

· 例：http://192.168.28.72:8161

activemq与spring整合

（5）、输入http://IP:8161/admin进行用户的登录，而后创建一个Queue，命名为FirstQueue。

5、Spring与activemq的整合

建立一个mq的web项目目录结构如下：

5.1在pom.xml文件中引入spring和mq的依赖

5.2配置生产者和消费者的xml文件

（1）配置spring-productorMQ.xml

[html]view
plain copy

<?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"  

    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.1.xsd   

        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd  

        http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.1.xsd">  

    <!-- 定义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>  

    <!-- 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>  

    <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->  

    <bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  

        <property name="brokerURL" value="tcp://192.168.154.158:61616"/>  

        <property name="transportListener"  ref="providerTransportListener"/>   

    </bean>  

    <!--Queue目的地-->  

    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  

        <constructor-arg name="name" value="FirstQueue"></constructor-arg>  

    </bean>   

    <!-- 配置MessageListenerContainer -->  

    <bean  

        class="org.springframework.jms.listener.DefaultMessageListenerContainer">  

        <property name="connectionFactory" ref="connectionFactory" />  

        <property name="destination" ref="queueDestination" />  

        <property name="messageListener" ref="myQueueListener" />  

    </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> -->  

    <!--Topic目的地-->  

    <!-- <bean id="itemTopic" class="org.apache.activemq.command.ActiveMQTopic">  

        <constructor-arg name="name" value="FirstTopic"></constructor-arg>  

    </bean> -->  

    <!-- 配置MessageListenerContainer -->  

    <!-- <bean  

        class="org.springframework.jms.listener.DefaultMessageListenerContainer">  

        <property name="connectionFactory" ref="connectionFactory" />  

        <property name="destination" ref="itemTopic" />  

        <property name="messageListener" ref="myTopicListener" />  

    </bean> -->  

    <!-- 自定义消息监听器，用于接收queue或topic之后做业务处理 -->  

    <!-- <bean id="myTopicListener" class="cn.fengfeng.activemq.consumer.TopicReceiver"></bean> -->  

</beans>

（2）配置spring-consumerMQ.xml

[html]view
plain copy

<?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"  

    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.1.xsd   

        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd  

        http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.1.xsd">  

    <!-- 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>  

    <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->  

    <bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  

        <property name="brokerURL" value="tcp://192.168.154.158:61616"/>  

        <property name="transportListener"  ref="providerTransportListener"/>   

    </bean>  

    <!--Queue目的地-->  

    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  

        <constructor-arg name="name" value="FirstQueue"></constructor-arg>  

    </bean>   

    <!-- mq监听类，对mq整体的变化的监控  -->  

    <bean id="providerTransportListener" class="cn.fengfeng.activemq.listener.ProviderTransportListener"/>  

    <!-- 自定义消息监听器，用于接收queue或topic之后做业务处理 -->  

    <bean id="myQueueListener" class="cn.fengfeng.activemq.consumer.Receiver"></bean>  

    <!-- 配置MessageListenerContainer监听容器 -->  

    <bean  

        class="org.springframework.jms.listener.DefaultMessageListenerContainer">  

        <property name="connectionFactory" ref="connectionFactory" />  

        <property name="destination" ref="queueDestination" />  

        <property name="messageListener" ref="myQueueListener" />  

    </bean>  

    <!--Topic目的地-->  

    <!-- <bean id="itemTopic" class="org.apache.activemq.command.ActiveMQTopic">  

        <constructor-arg name="name" value="FirstTopic"></constructor-arg>  

    </bean> -->  

    <!-- 配置MessageListenerContainer -->  

    <!-- <bean  

        class="org.springframework.jms.listener.DefaultMessageListenerContainer">  

        <property name="connectionFactory" ref="connectionFactory" />  

        <property name="destination" ref="itemTopic" />  

        <property name="messageListener" ref="myTopicListener" />  

    </bean> -->  

    <!-- 自定义消息监听器，用于接收queue或topic之后做业务处理 -->  

    <!-- <bean id="myTopicListener" class="cn.fengfeng.activemq.consumer.TopicReceiver"></bean> -->  

</beans>

（3）配置web.xml文件

[html]view
plain copy

<?xml version="1.0" encoding="UTF-8"?>  

<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://xmlns.jcp.org/xml/ns/javaee" xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd" id="WebApp_ID" version="3.1">  

  <display-name>mq</display-name>  

  <!-- 在WEB容器里面进行Spring容器的加载 -->  

    <listener>  

        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>  

    </listener>  

    <!-- 此配置描述的是在项目开发过程之中，Spring容器所需要使用到的配置文件 -->  

    <context-param>  

        <param-name>contextConfigLocation</param-name>  

        <param-value>  

            classpath*:spring-*.xml  

        </param-value>  

    </context-param>  

    <!-- 此为Spring MVC配置所需要的程序文件，所有的请求都提交给Spring的Servlet程序 -->  

    <servlet>  

        <servlet-name>springmvc</servlet-name>  

        <!-- 此为Spring MVC自己提供的servlet程序，一定要写上，因为其可以处理用户请求 -->  

        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>  

        <init-param>  

            <param-name>contextConfigLocation</param-name>  

            <!-- 配置Spring MVC所需要的配置文件 -->  

            <param-value>classpath:spring-mvc.xml</param-value>  

        </init-param>  

    </servlet>  

    <!-- SpringMVC中所有路径的请求映射，使用的是“*.action” -->  

    <servlet-mapping>  

        <servlet-name>springmvc</servlet-name>  

        <url-pattern>*.action</url-pattern>  

    </servlet-mapping>  

    <!-- Spring中提供的编码过滤器，使用的编码都是UTF-8 -->  

    <filter>  

        <filter-name>encoding</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>encoding</filter-name>  

        <url-pattern>/*</url-pattern>  

    </filter-mapping>  

  <welcome-file-list>  

    <welcome-file>index.html</welcome-file>  

    <welcome-file>index.htm</welcome-file>  

    <welcome-file>index.jsp</welcome-file>  

    <welcome-file>default.html</welcome-file>  

    <welcome-file>default.htm</welcome-file>  

    <welcome-file>default.jsp</welcome-file>  

  </welcome-file-list>  

</web-app>

5.4、生产者代码Sender.java

[java]view
plain copy

package cn.fengfeng.activemq.productor;  

import javax.annotation.Resource;  

import javax.jms.Destination;  

import javax.jms.JMSException;  

import javax.jms.Message;  

import javax.jms.Session;  

import org.springframework.jms.core.JmsTemplate;  

import org.springframework.jms.core.MessageCreator;  

import org.springframework.stereotype.Service;  

@Service  

public class Sender {  

    private static final int SEND_NUMBER = 5;  

    @Resource  

    private JmsTemplate jmsTemplate ;  

    //name属性用来明确使用哪个queue或者topic作为发送消息的Destination  

    @Resource(name = "queueDestination")   

    private Destination queueDestination ;  

    public void sendMessage(String message) throws InterruptedException {  

        for (int x = 0 ; x < SEND_NUMBER ; x ++ ) {  

            this.jmsTemplate.send(this.queueDestination, new MessageCreator() {  

                @Override  

                public Message createMessage(Session session) throws JMSException {  

                    System.out.println("***" + message + "***");  

                    return session.createTextMessage(message);  

                }  

            });  

            Thread.sleep(1000) ;  

        }  

    }  

}

5.5、消费者代码Receiver.java

[java]view
plain copy

package cn.fengfeng.activemq.consumer;  

import javax.jms.JMSException;  

import javax.jms.Message;  

import javax.jms.MessageListener;  

import javax.jms.TextMessage;  

public class Receiver implements MessageListener {  

    @Override  

    public void onMessage(Message message) {  

        if (message instanceof TextMessage ) {  

            TextMessage textMessage = (TextMessage) message;  

            try {  

                // 生产者发送消息后，就会被此类监听到，而后获取消息，进而做一些业务处理  

                System.out.println("###" + textMessage.getText() + "###");  

            } catch (JMSException e) {  

                // TODO Auto-generated catch block  

                e.printStackTrace();  

            }  

        }     

    }  

}

5.6、监听类

[java]view
plain copy

package cn.fengfeng.activemq.listener;  

import java.io.IOException;  

import org.apache.activemq.transport.TransportListener;  

import org.slf4j.Logger;  

import org.slf4j.LoggerFactory;  

public class ProviderTransportListener implements TransportListener {  

    private static final Logger Log = LoggerFactory.getLogger(ProviderTransportListener.class) ;  

    @Override  

    public void onCommand(Object arg0) {  

        // TODO Auto-generated method stub  

        Log.debug("监听到服务命令：{}" + arg0);  

    }  

    @Override  

    public void onException(IOException arg0) {  

        // TODO Auto-generated method stub  

        Log.debug("出现异常：{}" + arg0);  

    }  

    @Override  

    public void transportInterupted() {  

        // TODO Auto-generated method stub  

        Log.debug("传输中断。");  

    }  

    @Override  

    public void transportResumed() {  

        // TODO Auto-generated method stub  

        Log.debug("建立连接或传输恢复！");  

    }  

}

5.7、建立一个action

[java]view
plain copy

package cn.fengfeng.activemq.controller;  

import javax.annotation.Resource;  

import org.springframework.stereotype.Controller;  

import org.springframework.web.bind.annotation.RequestMapping;  

import org.springframework.web.servlet.ModelAndView;  

import cn.fengfeng.activemq.productor.Sender;  

@Controller  

@RequestMapping("/activemq/*")  

public class ActiveMQAction {  

    @Resource  

    private Sender sender ;  

    @RequestMapping("/send")  

    public ModelAndView send(String message) {  

        try {  

            this.sender.sendMessage(message);  

        } catch (InterruptedException e) {  

            e.printStackTrace();  

        }  

        return null ;  

    }  

}

将项目部署到tomcat下，先启动activemq的服务，而后启动tomcat，在浏览器中通过访问http://localhost/mq/activemq/send.action?message=helloworld 将会在后台中看到发送的消息和被监听到的消息如下图所示：

activemq与spring整合

相关推荐