spring整合activemq消息队列之点对点模式

关于jms概念以及activemq这里不做具体赘述，activemq是apache的一款项目，介绍里号称是最方便最强大的jms实现方式。其实jms与webservice功能是一样的，侧重点不同而已。都是提供多系统之间的交互通信方式，前者相对于后者更加轻量级，可以实现延迟通信，避免系统高峰期交互通信，也不需要暴露接口等，开发起来更加方便，但是如果项目要求即时性很高那么就选择后者，当然他们两者也可以整合起来一起使用，本文重点介绍的是spring框架整合activemq实现点对点模式，关于activemq具体概念百度。

1：下载activemq中间件，地址：http://activemq./ 

下载好压缩到本地，可以看到这个目录

从它的目录来说，还是很简单的：

• bin存放的是脚本文件
• conf存放的是基本配置文件
• data存放的是日志文件
• docs存放的是说明文档
• examples存放的是简单的实例
• lib存放的是activemq所需jar包
• webapps用于存放项目的目录

其中conf里面有个activemq.xml里面有个配置<broker xmlns="http://activemq./schema/core" brokerName="localhost" dataDirectory="${activemq.data}">默认是localhost，这个是可以修改，比如可以换成本机ip地址或是局域网另一台电脑ip,这样把中间件部署到同事电脑你也可以访问，同理部署到linux服务器上只需要把localhost换成服务器ip地址，关于这一段的修改用意在代码里面会提及。修改完后进入bin文件夹选择32或64位文件夹后点击activemq.bat批处理命令后中间件启动了起来。该项目还提供了一个管理页面，访问地址是http://(localhost或是你前面修改的同事ip或服务器ip地址):8161/admin   默认的用户名密码都是admin,见下图。我们主要点击Queues栏，里面会显示你的发送消息和接收消息的具体信息，也可以做删除等操作。

2：进行项目的搭建

pom.xml配置

[html] view plain copy

print?

1. <properties>  
2.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>  
3.     <springframework>4.1.8.RELEASE</springframework>  
4.   </properties>  
5.   
6.   <dependencies>  
7.     <dependency>  
8.       <groupId>junit</groupId>  
9.       <artifactId>junit</artifactId>  
10.       <version>3.8.1</version>  
11.       <scope>test</scope>  
12.     </dependency>  
13.       
14.     <dependency>  
15.       <groupId>javax.servlet</groupId>  
16.       <artifactId>jstl</artifactId>  
17.       <version>1.2</version>  
18.     </dependency>  
19.       
20.     <dependency>       
21.         <groupId>javax.servlet</groupId>  
22.         <artifactId>javax.servlet-api</artifactId>  
23.         <version>3.1.0</version>  
24.     </dependency>  
25.       
26.     <dependency>  
27.         <groupId>javax.servlet.jsp</groupId>  
28.         <artifactId>javax.servlet.jsp-api</artifactId>  
29.         <version>2.3.1</version>  
30.     </dependency>  
31.       
32.     <!-- spring -->  
33.     <dependency>  
34.         <groupId>org.springframework</groupId>  
35.         <artifactId>spring-core</artifactId>  
36.         <version>${springframework}</version>  
37.     </dependency>  
38.     <dependency>  
39.         <groupId>org.springframework</groupId>  
40.         <artifactId>spring-context</artifactId>  
41.         <version>${springframework}</version>  
42.     </dependency>  
43.     <dependency>  
44.         <groupId>org.springframework</groupId>  
45.         <artifactId>spring-tx</artifactId>  
46.         <version>${springframework}</version>  
47.     </dependency>  
48.     <dependency>  
49.         <groupId>org.springframework</groupId>  
50.         <artifactId>spring-webmvc</artifactId>  
51.         <version>${springframework}</version>  
52.     </dependency>  
53.     <dependency>  
54.         <groupId>org.springframework</groupId>  
55.         <artifactId>spring-web</artifactId>  
56.         <version>${springframework}</version>  
57.     </dependency>  
58.     <dependency>    
59.         <groupId>org.springframework</groupId>    
60.         <artifactId>spring-jms</artifactId>    
61.         <version>${springframework}</version>    
62.     </dependency>  
63.       
64.     <!-- xbean 如<amq:connectionFactory /> -->  
65.     <dependency>  
66.         <groupId>org.apache.xbean</groupId>  
67.         <artifactId>xbean-spring</artifactId>  
68.         <version>3.16</version>  
69.     </dependency>  
70.       
71.       
72.     <!-- activemq -->  
73.     <dependency>    
74.         <groupId>org.apache.activemq</groupId>    
75.         <artifactId>activemq-core</artifactId>    
76.         <version>5.7.0</version>  
77.     </dependency>   
78.     <dependency>  
79.         <groupId>org.apache.activemq</groupId>  
80.         <artifactId>activemq-pool</artifactId>  
81.         <version>5.12.1</version>  
82.     </dependency>  
83.       
84.     <!--实现日志  -->  
85.     <dependency>  
86.         <groupId>org.slf4j</groupId>  
87.         <artifactId>slf4j-log4j12</artifactId>  
88.         <version>1.7.22</version>  
89.     </dependency>  
90.       
91.     <!--处理json  -->  
92.     <dependency>  
93.             <groupId>com.alibaba</groupId>  
94.             <artifactId>fastjson</artifactId>  
95.             <version>1.2.0</version>  
96.     </dependency>  
97.   
98.   </dependencies>  

applicationContext.xml配置

[html] view plain copy

print?

1. <?xml version="1.0" encoding="UTF-8"?>  
2. <beans xmlns="http://www./schema/beans"  
3.     xmlns:xsi="http://www./2001/XMLSchema-instance"  
4.     xmlns:context="http://www./schema/context"  
5.     xmlns:mvc="http://www./schema/mvc"  
6.     xsi:schemaLocation="http://www./schema/beans   
7.         http://www./schema/beans/spring-beans.xsd  
8.         http://www./schema/context  
9.         http://www./schema/context/spring-context-4.1.xsd  
10.         http://www./schema/mvc   
11.         http://www./schema/mvc/spring-mvc-4.1.xsd">  
12.            
13.      <!-- 配置扫描路径 -->  
14.      <context:component-scan base-package="com.qqw">  
15.         <!-- 只扫描Service，也可以添加Repostory，但是要把Controller排除在外，Controller由spring-mvc.xml去加载 -->  
16.         <!-- <context:include-filter type="annotation" expression="org.springframework.stereotype.Service" /> -->  
17.         <!-- <context:include-filter type="annotation" expression="org.springframework.stereotype.Repository" /> -->  
18.         <!-- <context:include-filter type="annotation" expression="org.springframework.stereotype.Component" /> -->  
19.         <context:exclude-filter type="annotation" expression="org.springframework.stereotype.Controller"/>  
20.      </context:component-scan>  
21.     <!-- <mvc:annotation-driven /> -->  
22.        
23. </beans>  

spring-mvc.xml

[html] view plain copy

print?

1. <?xml version="1.0" encoding="UTF-8"?>  
2. <beans xmlns="http://www./schema/beans"  
3.     xmlns:xsi="http://www./2001/XMLSchema-instance"  
4.     xmlns:context="http://www./schema/context"  
5.     xmlns:mvc="http://www./schema/mvc"  
6.     xsi:schemaLocation="http://www./schema/beans   
7.         http://www./schema/beans/spring-beans.xsd  
8.         http://www./schema/context  
9.         http://www./schema/context/spring-context-4.1.xsd  
10.         http://www./schema/mvc   
11.         http://www./schema/mvc/spring-mvc-4.1.xsd">  
12.            
13.        
14.     <!-- 启用MVC注解 -->  
15.     <mvc:annotation-driven />  
16.       
17.     <!-- 静态资源文件，不会被Spring MVC拦截 -->  
18.     <mvc:resources location="/resources/" mapping="/resources/**"/>  
19.       
20.     <!-- 指定Sping组件扫描的基本包路径 -->  
21.     <context:component-scan base-package="com.qqw" >  
22.         <!-- 这里只扫描Controller，不可重复加载Service -->  
23.         <context:include-filter type="annotation" expression="org.springframework.stereotype.Controller"/>  
24.     </context:component-scan>  
25.       
26.     <!-- JSP视图解析器-->   
27.         <bean id="viewResolver" class="org.springframework.web.servlet.view.InternalResourceViewResolver">  
28.         <property name="viewClass" value="org.springframework.web.servlet.view.JstlView"/>  
29.         <!-- <property name="prefix" value="/WEB-INF/jsp/"/> -->  
30.         <property name="prefix" value="/"/>  
31.         <property name="suffix" value=".jsp"/>  
32.       
33.     <!--  定义其解析视图的order顺序为1 -->  
34.         <property name="order" value="1" />  
35.     </bean>  
36. </beans>  

接下来就是重要的spring-activemq.xml

[html] view plain copy

print?

1. <?xml version="1.0" encoding="UTF-8"?>  
2. <beans xmlns="http://www./schema/beans"  
3.     xmlns:xsi="http://www./2001/XMLSchema-instance"  
4.     xmlns:amq="http://activemq./schema/core"  
5.     xmlns:jms="http://www./schema/jms"  
6.     xmlns:context="http://www./schema/context"  
7.     xmlns:mvc="http://www./schema/mvc"  
8.     xsi:schemaLocation="  
9.         http://www./schema/beans       
10.         http://www./schema/beans/spring-beans-4.1.xsd  
11.         http://www./schema/context  
12.         http://www./schema/context/spring-context-4.1.xsd  
13.         http://www./schema/mvc  
14.         http://www./schema/mvc/spring-mvc-4.1.xsd  
15.         http://www./schema/jms  
16.         http://www./schema/jms/spring-jms-4.1.xsd  
17.         http://activemq./schema/core  
18.         http://activemq./schema/core/activemq-core-5.12.1.xsd"  
19.         >  
20.       
21.    <!--  <context:component-scan base-package="com.qqw" />  
22.     <mvc:annotation-driven />  
23.       
24.      -->  
25.    //这里的是用默认的localhost，如果做了修改或是项目上线，这个要换成网络ip地址，具体见上面的配置提到    
26.    <amq:connectionFactory id="amqConnectionFactory"   
27.         brokerURL="tcp://localhost:61616"   
28.         userName="admin"   
29.         password="admin" />  
30.       
31.     <!-- 配置JMS连接工长 -->  
32.     <bean id="connectionFactory"  
33.         class="org.springframework.jms.connection.CachingConnectionFactory">  
34.         <constructor-arg ref="amqConnectionFactory" />  
35.         <property name="sessionCacheSize" value="100" />  
36.     </bean>  
37.       
38.     <!-- 定义消息队列（Queue） -->  
39.     <bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
40.         <!-- 设置消息队列的名字 -->  
41.         <constructor-arg>  
42.             <value>gzframe.demo</value>  
43.         </constructor-arg>  
44.     </bean>  
45.       
46.     <!-- 配置JMS模板（Queue），Spring提供的JMS工具类，它发送、接收消息。 -->  
47.     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
48.         <property name="connectionFactory" ref="connectionFactory" />  
49.         <property name="defaultDestination" ref="demoQueueDestination" />  
50.         <property name="receiveTimeout" value="10000" />  
51.         <!-- true是topic，false是queue，默认是false，此处显示写出false -->  
52.         <property name="pubSubDomain" value="false" />  
53.     </bean>  
54.       
55.     <!--加入监听器  -->  
56.      <bean id="queueMessageListener" class="com.qqw.active.QueueMessageListener" />  
57.        
58.      <!-- 显示注入消息监听容器（Queue），配置连接工厂，监听的目标是demoQueueDestination，监听器是上面定义的监听器 -->  
59.     <bean id="queueListenerContainer"  
60.         class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
61.         <property name="connectionFactory" ref="connectionFactory" />  
62.         <property name="destination" ref="demoQueueDestination" />  
63.         <property name="messageListener" ref="queueMessageListener" />  
64.     </bean>  
65.  </beans>  

3：逻辑代码

     （1）消息发送者

[java] view plain copy

print?

1. package com.qqw.active;  
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.Service;  
12.   
13. @Service  
14. public class ProducerService {  
15.   
16.     @Resource(name="jmsTemplate")  
17.     private JmsTemplate jmsTemplate;  
18.          
19.       /** 
20.        * 向指定队列发送消息 
21.        */  
22.       public void sendMessage(Destination destination, final String msg) {  
23.         System.out.println("向队列" + destination.toString() + "发送了消息------------" + msg);  
24.         jmsTemplate.send(destination, new MessageCreator() {  
25.           public Message createMessage(Session session) throws JMSException {  
26.             return session.createTextMessage(msg);  
27.           }  
28.         });  
29.       }  
30.        
31.     /** 
32.      * 向默认队列发送消息 
33.      */  
34.       public void sendMessage(final String msg) {  
35.         String destination =  jmsTemplate.getDefaultDestination().toString();  
36.         System.out.println("向队列" +destination+ "发送了消息------------" + msg);  
37.         jmsTemplate.send(new MessageCreator() {  
38.           public Message createMessage(Session session) throws JMSException {  
39.             return session.createTextMessage(msg);  
40.           }  
41.         });  
42.        
43.       }  
44.       
45. }  

（2）消息接受者

[java] view plain copy

print?

1. package com.qqw.active;  
2.   
3. import javax.annotation.Resource;  
4. import javax.jms.Destination;  
5. import javax.jms.JMSException;  
6. import javax.jms.TextMessage;  
7.   
8. import org.springframework.jms.core.JmsTemplate;  
9. import org.springframework.stereotype.Service;  
10.   
11. import com.alibaba.fastjson.JSON;  
12.   
13.   
14. @Service  
15. public class ConsumerService {  
16.   
17.      @Resource(name="jmsTemplate")  
18.         private JmsTemplate jmsTemplate;  
19.        
20.     /** 
21.      * 接收消息 
22.      */  
23.     public TextMessage receive(Destination destination) {  
24.         TextMessage tm = (TextMessage) jmsTemplate.receive(destination);  
25.         try {  
26.             System.out.println("从队列" + destination.toString() + "收到了消息：\t"  
27.                     + tm.getText());  
28.             /*Gson gson = new Gson();*/  
29.             Person fromJson = JSON.parseObject(tm.getText(), Person.class);  
30.             System.out.println(fromJson.toString());  
31.         } catch (JMSException e) {  
32.             e.printStackTrace();  
33.         }  
34.           
35.         return tm;  
36.           
37.     }  
38.    
39. }  

（3）传递的实体类

[java] view plain copy

print?

1. package com.qqw.active;  
2.   
3. public class Person {  
4.   
5.     private String name;  
6.     private String sex;  
7.     private int age;  
8.     public Person() {  
9.         super();  
10.         // TODO Auto-generated constructor stub  
11.     }  
12.     public String getName() {  
13.         return name;  
14.     }  
15.     public void setName(String name) {  
16.         this.name = name;  
17.     }  
18.     public String getSex() {  
19.         return sex;  
20.     }  
21.     public void setSex(String sex) {  
22.         this.sex = sex;  
23.     }  
24.     public int getAge() {  
25.         return age;  
26.     }  
27.     public void setAge(int age) {  
28.         this.age = age;  
29.     }  
30.     @Override  
31.     public String toString() {  
32.         return "Person [name=" + name + ", sex=" + sex + ", age=" + age + "]";  
33.     }  
34.       
35.       
36. }  

（4）为接受者定义监听器，这样消息一发到中间件会自动接收处理，如果不需要自动，这段不用写，并且把activemq.xml后两个bean注释掉

[java] view plain copy

print?

1. package com.qqw.active;  
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 QueueMessageListener implements MessageListener {  
9.   
10.       
11.     //当收到消息后，自动调用该方法  
12.     @Override  
13.     public void onMessage(Message message) {  
14.           
15.         TextMessage tm = (TextMessage) message;  
16.         try {  
17.             System.out.println("QueueMessageListener监听到了文本消息：\t"  
18.                     + tm.getText());            
19.   
20.             Person fromJson = JSON.parseObject(tm.getText(), Person.class);  
21.             System.out.println(fromJson.toString());  
22.   
23.            //do something ...  
24.         } catch (JMSException e){ e.printStackTrace();   
25.         }   
26.         }  
27.     }  
28.   
29. }  

（5）具体业务逻辑

[java] view plain copy

print?

1. //发送queue消息  
2.     @RequestMapping(value="/onsend",method=RequestMethod.POST)  
3.     public ModelAndView producer(@RequestParam("message") String message) {  
4.         System.out.println("------------send to jms");  
5.         com.alibaba.fastjson.JSONObject jsonObj = new com.alibaba.fastjson.JSONObject();  
6.         jsonObj.put("name", "张三");  
7.         jsonObj.put("sex", "男");  
8.         jsonObj.put("age", 20);  
9.         String resultjson=JSON.toJSONString(jsonObj);  
10.         ModelAndView mv = null;  
11.         producer.sendMessage(demoQueueDestination, resultjson);  
12.         mv=new ModelAndView("index");  
13.         return mv;  
14.     }  
15.       
16.     //如果不需要监听器想手动去接收消息的方式  
17.     @RequestMapping(value="/receive",method=RequestMethod.GET)  
18.     public ModelAndView queue_receive() throws JMSException {  
19.         System.out.println("------------receive message");  
20.         ModelAndView mv = new ModelAndView();  
21.           
22.         TextMessage tm = consumer.receive(demoQueueDestination);  
23.         mv.addObject("textMessage", tm.getText());  
24.           
25.         mv.setViewName("queue_receive");  
26.         return mv;  
27.     }  

(6)本地运行截图

以上就介绍完了，参考了网上资料以及自己的一些实践，大家互相交流学习！！