Spring和ActiveMQ集成实现队列消息以及PUB/SUB模型

2019年11月11日 阅读数:167
这篇文章主要向大家介绍Spring和ActiveMQ集成实现队列消息以及PUB/SUB模型,主要内容包括基础应用、实用技巧、原理机制等方面,希望对大家有所帮助。

前言:本文是基于Spring和ActiveMQ的一个示例文章,包括了Point-To-Point的异步队列消息和PUB/SUB(发布/订阅)模型,只是作了比较简单的实现,无任何业务方面的东西,做为一个入门教程。javascript

适合对象:但愿学习ActiveMQ的朋友,以及利用Spring将ActiveMQ集成到系统中css

目录结构html

Maven依赖:前端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
<!-- Jar版本管理 -->
     < properties >
         < springframework >4.0.2.RELEASE</ springframework >
         < log4j >1.2.17</ log4j >
         < activemq >5.9.0</ activemq >
     </ properties >
 
     < dependencies >
         <!-- Spring web mvc -->
         < dependency >
             < groupId >org.springframework</ groupId >
             < artifactId >spring-webmvc</ artifactId >
             < version >${springframework}</ version >
         </ dependency >
 
         <!-- 提供JMS,Freemarker,Quartz集成服务 -->
         < dependency >
             < groupId >org.springframework</ groupId >
             < artifactId >spring-context-support</ artifactId >
             < version >${springframework}</ version >
         </ dependency >
         
         <!-- 集成JMS -->
         < dependency >
             < groupId >org.springframework</ groupId >
             < artifactId >spring-jms</ artifactId >
             < version >${springframework}</ version >
         </ dependency >
         
         <!-- xbean 如<amq:connectionFactory /> -->
         < dependency >
             < groupId >org.apache.xbean</ groupId >
             < artifactId >xbean-spring</ artifactId >
             < version >3.16</ version >
         </ dependency >
 
         <!-- log4j -->
         < dependency >
             < groupId >log4j</ groupId >
             < artifactId >log4j</ artifactId >
             < version >${log4j}</ version >
         </ dependency >
 
         <!-- Active MQ -->
         < dependency >
             < groupId >org.apache.activemq</ groupId >
             < artifactId >activemq-all</ artifactId >
             < version >${activemq}</ version >
         </ dependency >
 
         <!-- 单元测试 -->
         < dependency >
             < groupId >junit</ groupId >
             < artifactId >junit</ artifactId >
             < version >3.8.1</ version >
             < scope >test</ scope >
         </ dependency >
     </ dependencies >

 

jar包截图java

web.xmljquery

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
<? xml  version = "1.0"  encoding = "UTF-8" ?>
<!-- 经过http://java.sun.com/xml/ns/javaee/获取最新的schemaLocation -->
< web-app  xmlns:xsi = "http://www.w3.org/2001/XMLSchema-instance"
     xmlns = "http://java.sun.com/xml/ns/javaee"  xmlns:web = "http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
     xsi:schemaLocation = "http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
     version = "3.0" >
     < display-name >SpringActivemqServer</ display-name >
 
     <!-- WebAppRootKey -->
     < context-param >
         < param-name >webAppRootKey</ param-name >
         < param-value >example.SpringActivemqServer</ param-value >
     </ context-param >
 
     <!-- Log4J Start -->
     < context-param >
         < param-name >log4jConfigLocation</ param-name >
         < param-value >classpath:log4j.properties</ param-value >
     </ context-param >
     < context-param >
         < param-name >log4jRefreshInterval</ param-name >
         < param-value >6000</ param-value >
     </ context-param >
     <!-- Spring Log4J config -->
     < listener >
         < listener-class >org.springframework.web.util.Log4jConfigListener</ listener-class >
     </ listener >
     <!-- Log4J End -->
 
     <!-- Spring 编码过滤器 start -->
     < filter >
         < filter-name >characterEncoding</ 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 >
         < init-param >
             < param-name >forceEncoding</ param-name >
             < param-value >true</ param-value >
         </ init-param >
     </ filter >
     < filter-mapping >
         < filter-name >characterEncoding</ filter-name >
         < url-pattern >/*</ url-pattern >
     </ filter-mapping >
     <!-- Spring 编码过滤器 End -->
 
     <!-- Spring Application Context Listener Start -->
     < context-param >
         < param-name >contextConfigLocation</ param-name >
         < param-value >classpath*:applicationContext.xml,classpath*:ActiveMQ.xml</ param-value >
     </ context-param >
     < listener >
         < listener-class >org.springframework.web.context.ContextLoaderListener</ listener-class >
     </ listener >
     <!-- Spring Application Context Listener End -->
 
 
     <!-- Spring MVC Config Start -->
     < servlet >
         < servlet-name >SpringMVC</ 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 >SpringMVC</ servlet-name >
         <!-- Filter all resources -->
         < url-pattern >/</ url-pattern >
     </ servlet-mapping >
     <!-- Spring MVC Config End -->
 
</ web-app >

applicationContext.xmlgit

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
<? xml  version = "1.0"  encoding = "UTF-8" ?>
<!-- 查找最新的schemaLocation 访问 http://www.springframework.org/schema/ -->
< 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:amq = "http://activemq.apache.org/schema/core"
     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
         http://activemq.apache.org/schema/core
         http://activemq.apache.org/schema/core/activemq-core-5.9.0.xsd">
      
      < bean  class = "org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor" />
      
      <!-- 配置扫描路径 -->
      < context:component-scan  base-package = "org.xdemo.example" >
        <!-- 只扫描Service,也能够添加Repostory,可是要把Controller排除在外,Controller由spring-mvc.xml去加载 -->
        <!-- <context:include-filter type="annotation" expression="org.springframework.stereotype.Service" /> -->
        <!-- <context:include-filter type="annotation" expression="org.springframework.stereotype.Repository" /> -->
        <!-- <context:include-filter type="annotation" expression="org.springframework.stereotype.Component" /> -->
        < context:exclude-filter  type = "annotation"  expression = "org.springframework.stereotype.Controller" />
      </ context:component-scan >
 
</ beans >

spring-mvc.xmlweb

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
<? xml  version = "1.0"  encoding = "UTF-8" ?>  
<!-- 查找最新的schemaLocation 访问 http://www.springframework.org/schema/ -->
< beans  xmlns = "http://www.springframework.org/schema/beans"   
        xmlns:aop = "http://www.springframework.org/schema/aop"   
        xmlns:context = "http://www.springframework.org/schema/context"  
        xmlns:mvc = "http://www.springframework.org/schema/mvc"   
        xmlns:tx = "http://www.springframework.org/schema/tx"   
        xmlns:xsi = "http://www.w3.org/2001/XMLSchema-instance"  
        xsi:schemaLocation="http://www.springframework.org/schema/aop   
         http://www.springframework.org/schema/aop/spring-aop-4.0.xsd   
         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   
         http://www.springframework.org/schema/mvc   
         http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd   
         http://www.springframework.org/schema/tx   
         http://www.springframework.org/schema/tx/spring-tx-4.0.xsd">  
   
       <!-- 启用MVC注解 -->
     < mvc:annotation-driven  />
     
     <!-- 静态资源文件,不会被Spring MVC拦截 -->
     < mvc:resources  location = "/resources/"  mapping = "/resources/**" />
     
     <!-- 指定Sping组件扫描的基本包路径 -->
     < context:component-scan  base-package = "org.xdemo.example"  >
         <!-- 这里只扫描Controller,不可重复加载Service -->
         < context:include-filter  type = "annotation"  expression = "org.springframework.stereotype.Controller" />
     </ context:component-scan >
     
       <!-- JSP视图解析器-->
     < 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 >

ActiveMQ.xmlajax

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
<? xml  version = "1.0"  encoding = "UTF-8" ?>
<!-- 查找最新的schemaLocation 访问 http://www.springframework.org/schema/ -->
< 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: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/context   
         http://www.springframework.org/schema/context/spring-context-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://localhost:61616"  userName = "admin"  password = "admin"  />
 
     < bean  id = "connectionFactory"
         class = "org.springframework.jms.connection.CachingConnectionFactory" >
         < constructor-arg  ref = "amqConnectionFactory"  />
         < property  name = "sessionCacheSize"  value = "100"  />
     </ bean >
     
     <!-- ====Producer side start====-->
     
     <!-- 定义JmsTemplate的Queue类型 -->
     < bean  id = "jmsQueueTemplate"  class = "org.springframework.jms.core.JmsTemplate" >
         < constructor-arg  ref = "connectionFactory"  />
         <!-- 非pub/sub模型(发布/订阅),即队列模式 -->
         < property  name = "pubSubDomain"  value = "false"  />
     </ bean >
     
     <!-- 定义JmsTemplate的Topic类型 -->
     < bean  id = "jmsTopicTemplate"  class = "org.springframework.jms.core.JmsTemplate" >
         < constructor-arg  ref = "connectionFactory"  />
         <!-- pub/sub模型(发布/订阅) -->
         < property  name = "pubSubDomain"  value = "true"  />
     </ bean >
     
     <!-- ====Producer side end====-->
 
     
     <!-- ====Consumer side start====-->
     
     <!-- 定义Queue监听器 -->
     < jms:listener-container  destination-type = "queue"  container-type = "default"  connection-factory = "connectionFactory"  acknowledge = "auto" >
         < jms:listener  destination = "test.queue"  ref = "queueReceiver" />
         < 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 = "test.topic"  ref = "topicReceiver" />
         < jms:listener  destination = "test.topic"  ref = "topicReceiver2" />
     </ jms:listener-container >
     
     <!-- ====Consumer side end==== -->
</ beans >

队列消息生产者spring

QueueSender.java

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
package  org.xdemo.example.SpringActivemq.mq.producer.queue;
 
import  javax.jms.JMSException;
import  javax.jms.Message;
import  javax.jms.Session;
 
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;
 
/**
  * @做者 Goofy
  * @邮件 252878950@qq.com
  * @日期 2014-4-1上午9:40:24
  * @描述 发送消息到队列
  */
@Component
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);
             }
         });
     }
     
}

主题(Topic)消息生产者TopicSender.java

1
2
3
4
5