重庆分公司
重庆分公司，新征程启航
为企业提供网站建设、域名注册、服务器等服务
实战Spring4+ActiveMQ整合实现消息队列(生产者+消费者)
引言:
最近公司做了一个以信息安全为主的项目，其中有一个业务需求就是，项目定时监控操作用户的行为，对于一些违规操作严重的行为，以发送邮件(FoxMail)的形式进行邮件告警，可能是多人，也可能是一个人，第一次是以单人的形式，,直接在业务层需要告警的地方发送邮件即可，可是后边需求变更了，对于某些告警邮件可能会发送多人，这其中可能就会有阻塞发邮件的可能，直到把所有邮件发送完毕后再继续做下边的业务，领导说这样会影响用户体验，发邮件的时候用户一直处于等待状态，不能干别的事情。最后研究说用消息队列，当有需要发送邮件告警的时候,就向队列中添加一个标识消息，ActiveMQ通过监听器的形式，实时监听队列里边的小时，收到消息后，判断是不是需要发送告警的标识，是的话就自行就行发送邮件！这是就研究的消息队列ActiveMQ,下边就是具体内容：
在辛集等地区，都构建了全面的区域性战略布局，加强发展的系统性、市场前瞻性、产品创新能力，以专注、极致的服务理念，为客户提供成都网站设计、做网站、成都外贸网站建设公司 网站设计制作按需网站制作,公司网站建设,企业网站建设,成都品牌网站建设,成都全网营销推广,成都外贸网站建设公司,辛集网站建设费用合理。
一、ActiveMQ
1.1). ActiveMQ
ActiveMQ是Apache所提供的一个开源的消息系统，完全采用Java来实现，因此，它能很好地支持J2EE提出的JMS（Java Message Service,即Java消息服务）规范。JMS是一组Java应用程序接口，它提供消息的创建、发送、读取等一系列服务。JMS提供了一组公共应用程序接口和响应的语法，类似于Java数据库的统一访问接口JDBC,它是一种与厂商无关的API，使得Java程序能够与不同厂商的消息组件很好地进行通信。
1. 2). Java Message Service(JMS)
JMS支持两种消息发送和接收模型。
- 一种称为P2P(Ponit to Point)模型(点对点一对一)，即采用点对点的方式发送消息。P2P模型是基于队列的，消息生产者发送消息到队列，消息消费者从队列中接收消息，队列的存在使得消息的异步传输称为可能，P2P模型在点对点的情况下进行消息传递时采用。
- 另一种称为Pub/Sub(Publish/Subscribe，即发布-订阅)模型，发布-订阅模型定义了如何向一个内容节点发布和订阅消息，这个内容节点称为topic(主题)。主题可以认为是消息传递的中介，消息发布这将消息发布到某个主题，而消息订阅者则从主题订阅消息。主题使得消息的订阅者与消息的发布者互相保持独立，不需要进行接触即可保证消息的传递，发布-订阅模型在消息的一对多广播时采用。
1.3). JMS术语
- Provider/MessageProvider：生产者
- Consumer/MessageConsumer：消费者
- PTP：Point To Point，点对点通信消息模型
- Pub/Sub：Publish/Subscribe，发布订阅消息模型
- Queue：队列，目标类型之一，和PTP结合
- Topic：主题，目标类型之一，和Pub/Sub结合
- ConnectionFactory：连接工厂，JMS用它创建连接
- Connnection：JMS Client到JMS Provider的连接
- Destination：消息目的地，由Session创建
- Session：会话，由Connection创建，实质上就是发送、接受消息的一个线程，因此生产者、消费者都是 Session创建的
1.4). ActiveMQ应用场景
类似送快递，快递员（producer）将快递（Message）放到指定地点（destination）后，就可以离开了，拿快递的人（customer）在接收到通知后，到指定地点（destination）去取快递（Message）就可以了。当然，取快递时可能要进行身份验证，这就涉及到创建连接（connection）时，需要指定用户名和密码了。还有就是，实际生活中，当快递员把快递放好之后，照理说应该通知客户去哪里取快递，而ActiveMq帮我们做好了一切，通知的工作Activemq会帮我们实现，而无需我们亲自编码通知消费者，生产者只需要将Message放到Mq中即可，通知消费者的工作，mq会帮我们处理
用途就是用来处理消息，也就是处理JMS在大型电子商务类网站，如京东、淘宝、去哪儿等网站有着深入的应用，队列的主要作用是消除高并发访问高峰，加快网站的响应速度。
在不使用消息队列的情况下，用户的请求数据直接写入数据库，高发的情况下，会对数据库造成巨大的压力，同时也使得系统响应延迟加剧，但使用队列后，用户的请求发给队列后立即返回。
1.5). ActiveMQ下载
1.6). 启动
/apache-activemq-5.15.3/bin/win64/目录下双击activemq.bat文件，在浏览器中输入http://localhost:8161/admin/, 用户名和密码输入admin即可
二、Srping+ActiveMQ应用实例
2,1). 项目结构
2,2). 导入maven依赖,pom.xml文件
1
2
4 4.0.0
5
6 www.cnblogs.com.hongmoshu
7 test_actmq
8 0.0.1-SNAPSHOT
9 war
10 test_actmq Maven Webapp
11 http://www.example.com
12
13
14
15 4.1.8.RELEASE
16
17
18
19
20
21
22
23 junit
24 junit
25 4.11
26 test
27
28
29
30
31 jstl
32 jstl
33 1.2
34
35
36 javax.servlet
37 servlet-api
38 provided
39 2.5
40
41
42
43
44 org.springframework
45 spring-core
46 ${springframework}
47
48
49 org.springframework
50 spring-context
51 ${springframework}
52
53
54 org.springframework
55 spring-tx
56 ${springframework}
57
58
59 org.springframework
60 spring-webmvc
61 ${springframework}
62
63
64 org.springframework
65 spring-jms
66 ${springframework}
67
68
69
70
71 org.apache.xbean
72 xbean-spring
73 3.16
74
75
76
77
78 org.apache.activemq
79 activemq-core
80 5.7.0
81
82
83 org.apache.activemq
84 activemq-pool
85 5.12.1
86
87
88
89
90 com.google.code.gson
91 gson
92 1.7.1
93
94
95
96
97 net.sf.json-lib
98 json-lib
99 2.4
100 jdk15
101
102
103
104
105
106 test_actmq
107
108
109 2,3). ActiveMQ的配置文件ActiveMQ.xml
1
2
20
21
33
36
37
38
39
40
41 Jaycekon
42
43
44
45
46
47
55
56
57
66
70
71 2,4). Spring的配置文件 spring-mvc.xml
1
2
12
13
16
21
22 2,5). web.xml
1
2
6 mydemo
7
8
9 index.jsp
10
11
12
13
14 contextConfigLocation
15
16 classpath:spring-mvc.xml;
17 classpath:ActiveMQ.xml;
18
19
20
21
22
23 org.springframework.web.context.ContextLoaderListener
24
25
26
27 springMVC
28 org.springframework.web.servlet.DispatcherServlet
29
30 contextConfigLocation
31 classpath:spring-mvc.xml
32
33 1
34
35
36 springMVC
37 /
38
39
40
41
42 characterEncodingFilter
43 org.springframework.web.filter.CharacterEncodingFilter
44
45 encoding
46 UTF-8
47
48
49 forceEncoding
50 true
51
52
53
54 characterEncodingFilter
55 /*
56
57
58 2,6). 实体类Users对象
1 package com.svse.entity;
2 import java.io.Serializable;
3
4 public class Users implements Serializable{
5
6 private String userId;
7 private String userName;
8 private String sex;
9 private String age;
10 private String type;
11
12
13 public Users() {
14 super();
15 }
16 public Users(String userId, String userName, String sex, String age,
17 String type) {
18 super();
19 this.userId = userId;
20 this.userName = userName;
21 this.sex = sex;
22 this.age = age;
23 this.type = type;
24 }
25 public String getUserId() {
26 return userId;
27 }
28 public void setUserId(String userId) {
29 this.userId = userId;
30 }
31 public String getUserName() {
32 return userName;
33 }
34 public void setUserName(String userName) {
35 this.userName = userName;
36 }
37 public String getSex() {
38 return sex;
39 }
40 public void setSex(String sex) {
41 this.sex = sex;
42 }
43 public String getAge() {
44 return age;
45 }
46 public void setAge(String age) {
47 this.age = age;
48 }
49 public String getType() {
50 return type;
51 }
52 public void setType(String type) {
53 this.type = type;
54 }
55 @Override
56 public String toString() {
57 return "Users [userId=" + userId + ", userName=" + userName + ", sex="
58 + sex + ", age=" + age + ", type=" + type + "]";
59 }
60
61
62 }2,7). 核心代码(生产者ProducerService)
1 package com.svse.service;
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 import com.svse.entity.Users;
14
15 @Service
16 public class ProducerService {
17
18 @Resource(name="jmsTemplate")
19 private JmsTemplate jmsTemplate;
20
21
22 /**
23 * 向指定队列发送消息 (发送文本消息)
24 */
25 public void sendMessage(Destination destination,final String msg){
26
27 jmsTemplate.setDeliveryPersistent(true);
28
29 System.out.println(Thread.currentThread().getName()+" 向队列"+destination.toString()+"发送消息---------------------->"+msg);
30 jmsTemplate.send(destination, new MessageCreator() {
31 public Message createMessage(Session session) throws JMSException {
32 return session.createTextMessage(msg);
33 }
34 });
35 }
36
37 /**
38 * 向指定队列发送消息以对象的方式 (发送对象消息)
39 */
40 public void sendMessageNew(Destination destination,Users user){
41 System.out.println(Thread.currentThread().getName()+" 向队列"+destination.toString()+"发送消息---------------------->"+user);
42 jmsTemplate.convertAndSend(user);
43 }
44
45 /**
46 * 向默认队列发送消息
47 */
48 public void sendMessage(final String msg){
49 String destination = jmsTemplate.getDefaultDestinationName();
50 System.out.println(Thread.currentThread().getName()+" 向队列"+destination+"发送消息---------------------->"+msg);
51 jmsTemplate.send(new MessageCreator() {
52 public Message createMessage(Session session) throws JMSException {
53 return session.createTextMessage(msg);
54 }
55 });
56 }
57 }2,8). 核心代码(消费产者ConsumerService)
1 package com.svse.service;
2
3 import javax.annotation.Resource;
4 import javax.jms.Destination;
5 import javax.jms.JMSException;
6 import javax.jms.ObjectMessage;
7 import javax.jms.TextMessage;
8
9 import net.sf.json.JSONObject;
10
11 import org.springframework.jms.core.JmsTemplate;
12 import org.springframework.stereotype.Service;
13
14 import com.svse.entity.Users;
15
16 @Service
17 public class ConsumerService {
18
19 @Resource(name="jmsTemplate")
20 private JmsTemplate jmsTemplate;
21 //接收文本消息
22 public TextMessage receive(Destination destination){
23 TextMessage textMessage = (TextMessage) jmsTemplate.receive(destination);
24 try{
25 JSONObject json=JSONObject.fromObject(textMessage.getText());
26 System.out.println("name:"+json.getString("userName"));
27 System.out.println("从队列" + destination.toString() + "收到了消息：\t"
28 + textMessage.getText());
29 } catch (JMSException e) {
30 e.printStackTrace();
31 }
32 return textMessage;
33 }
34 //接收对象消息
35 public ObjectMessage receiveNew(Destination destination){
36 ObjectMessage objMsg=(ObjectMessage) jmsTemplate.receive(destination);
38 try{
39 Users users= (Users) objMsg.getObject();
44 System.out.println("name:"+users.getUserName());
47 System.out.println("从队列" + destination.toString() + "收到了消息：\t"
48 + users);
49 } catch (JMSException e) {
50 e.printStackTrace();
51 }
52 return objMsg;
53 }
54 }2,9). 核心代码(控制器ConsumerService)
1 package com.svse.controller.mq;
2
3 import java.io.IOException;
4 import java.text.SimpleDateFormat;
5 import java.util.Date;
7 import javax.annotation.Resource;
8 import javax.jms.DeliveryMode;
9 import javax.jms.Destination;
10 import javax.jms.JMSException;
11 import javax.jms.ObjectMessage;
12 import javax.jms.TextMessage;
13 import javax.management.MBeanServerConnection;
14 import javax.management.remote.JMXConnector;
15 import javax.management.remote.JMXConnectorFactory;
16 import javax.management.remote.JMXServiceURL;
18 import org.springframework.stereotype.Controller;
19 import org.springframework.web.bind.annotation.RequestMapping;
20 import org.springframework.web.bind.annotation.RequestMethod;
21 import org.springframework.web.bind.annotation.RequestParam;
22 import org.springframework.web.servlet.ModelAndView;
24 import com.google.gson.Gson;
25 import com.svse.entity.Users;
26 import com.svse.service.ConsumerService;
27 import com.svse.service.ProducerService;
28
29 @Controller
30 public class DemoController {
35
36 //队列名Jaycekon (ActiveMQ中设置的队列的名称)
37 @Resource(name="demoQueueDestination")
38 private Destination demoQueueDestination;
39
40 //队列消息生产者
41 @Resource(name="producerService")
42 private ProducerService producer;
43
44 //队列消息消费者
45 @Resource(name="consumerService")
46 private ConsumerService consumer;
47
48 /*
49 * 准备发消息
50 */
51 @RequestMapping(value="/producer",method=RequestMethod.GET)
52 public ModelAndView producer(){
53 System.out.println("------------go producer");
54
55 Date now = new Date();
56 SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
57 String time = dateFormat.format( now );
58 System.out.println(time);
59
60 ModelAndView mv = new ModelAndView();
61 mv.addObject("time", time);
62 mv.setViewName("producer");
63 return mv;
64 }
65
66 /*
67 * 发消息
68 */
69 @RequestMapping(value="/onsend",method=RequestMethod.POST)
70 public ModelAndView producer(@RequestParam("message") String message) {
71 System.out.println("------------send to jms");
72 ModelAndView mv = new ModelAndView();
73 for(int i=0;i<5;i++){
74 try {
75 Users users=new Users("10"+(i+1),"赵媛媛"+(i+1),"女","27","影视演员");
76 Gson gson=new Gson();
77 String sendMessage=gson.toJson(users);
78 System.out.println("发送的消息sendMessage:"+sendMessage.toString());
79 // producer.sendMessage(demoQueueDestination,sendMessage.toString());//以文本的形式
80 producer.sendMessageNew(demoQueueDestination, users);//以对象的方式
81
82 } catch (Exception e) {
83 e.printStackTrace();
84 }
85 }
86 mv.setViewName("index");
87 return mv;
88 }
89 /*
90 * 手动接收消息
91 */
92 @RequestMapping(value="/receive",method=RequestMethod.GET)
93 public ModelAndView queue_receive() throws JMSException {
94 System.out.println("------------receive message");
95 ModelAndView mv = new ModelAndView();
96
97 //TextMessage tm = consumer.receive(demoQueueDestination);//接收文本消息
98
99 ObjectMessage objMsg=consumer.receiveNew(demoQueueDestination);//接收对象消息
100 Users users= (Users) objMsg.getObject();
101 //mv.addObject("textMessage", tm.getText());
102 mv.addObject("textMessage", users.getUserId()+" || "+users.getUserName());
103 mv.setViewName("receive");
104 return mv;
105 }
106
107 /*
108 * ActiveMQ Manager Test
109 */
110 @RequestMapping(value="/jms",method=RequestMethod.GET)
111 public ModelAndView jmsManager() throws IOException {
112 System.out.println("------------jms manager");
113 ModelAndView mv = new ModelAndView();
114 mv.setViewName("index");
115
116 JMXServiceURL url = new JMXServiceURL("");
117 JMXConnector connector = JMXConnectorFactory.connect(url);
118 connector.connect();
119 MBeanServerConnection connection = connector.getMBeanServerConnection();
120
121 return mv;
122 }
123
124 }三、.对象转换器MessageConverter和消息监听器MessageListener
在上边的ProducerService和ConsumerService中,不论是发送消息还是接收消息,都可以以文本TextMessage的方式和ObjectMessage的方式.如果是简单的文本消息可以以TextMessage,但是如果需要发送的内容比较多,结构比较复杂,这时候就建议用对象文本ObjectMessage的方式向队列queue中发送消息了.但是这时候就需要用到对象消息转换器MessageConverter.
3,1). 消息转换器MessageageConverte
MessageConverter的作用主要有两方面，一方面它可以把我们的非标准化Message对象转换成我们的目标Message对象，这主要是用在发送消息的时候；另一方面它又可以把我们的Message对象
转换成对应的目标对象，这主要是用在接收消息的时候。
1 package com.svse.util;
2
3 import java.io.Serializable;
4
5 import javax.jms.JMSException;
6 import javax.jms.Message;
7 import javax.jms.ObjectMessage;
8 import javax.jms.Session;
9
10 import org.springframework.jms.support.converter.MessageConversionException;
11 import org.springframework.jms.support.converter.MessageConverter;
12
13 /**
14 *功能说明:通用的消息对象转换类
15 *@author:zsq
16 *create date:2019年7月12日 上午9:28:31
17 *修改人 修改时间 修改描述
18 *Copyright (c)2019北京智华天成科技有限公司-版权所有
19 */
20 public class ObjectMessageConverter implements MessageConverter {
21
22
23 //把一个Java对象转换成对应的JMS Message (生产消息的时候)
24 public Message toMessage(Object object, Session session)
25 throws JMSException, MessageConversionException {
26
27 return session.createObjectMessage((Serializable) object);
28 }
29
30 //把一个JMS Message转换成对应的Java对象 (消费消息的时候)
31 public Object fromMessage(Message message) throws JMSException,
32 MessageConversionException {
33 ObjectMessage objMessage = (ObjectMessage) message;
34 return objMessage.getObject();
35 }
36
37 }注意:写了消息转化器之后还需要的ActiveMQ.xml中进行配置
3,2). 消息监听器MessageageListe
MessageageListe作用就是动态的自行监听消息队列的生产者发送的消息,不需要人工手动接收！
1 package com.svse.util;
2 import javax.jms.JMSException;
3 import javax.jms.Message;
4 import javax.jms.MessageListener;
5 import javax.jms.ObjectMessage;
6 import javax.jms.TextMessage;
7
8 import com.svse.entity.Users;
9
10
11 public class QueueMessageListener implements MessageListener {
12
13 //添加了监听器,只要生产者发布了消息,监听器监听到有消息消费者就会自动消费(获取消息)
14 public void onMessage(Message message) {
15 //(第1种方式)没加转换器之前接收到的是文本消息
16 //TextMessage tm = (TextMessage) message;
17
18 //(第2种方式)加了转换器之后接收到的ObjectMessage对象消息
19 ObjectMessage objMsg=(ObjectMessage) message;
20 Users users;
21 try {
22 users = (Users) objMsg.getObject();
23 //System.out.println("QueueMessageListener监听到了文本消息：\t" + tm.getText());
24 System.out.println("QueueMessageListener监听到了文本消息：\t" + users);
25 //do something ...
26 } catch (JMSException e1) {
27 // TODO Auto-generated catch block
28 e1.printStackTrace();
29 }
30 }
31
32 }同样写好监听器后也是需在ActiveMQ.xml中进行配置注册的
总结
（1）注册JmsTemplate时，pubSubDomain这个属性的值要特别注意。默认值是false，也就是说默认只是支持queue模式，不支持topic模式。但是，如果将它改为true，则不支持queue模式。因此如果项目需要同时支持queue和topic模式，那么需要注册2个JmsTemplate，同时监听容器也需要注册2个
（2）使用Queue时，生产者只要将Message发送到MQ服务器端，消费者就可以进行消费，而无需生产者程序一直运行；
（3）消息是按照先入先出的顺序，一旦有消费者将Message消费，该Message就会从MQ服务器队列中删去；
（4）有文章说，“生产者”<-->"消费者"是一对一的关系，其实并不准确，从应用中可以看出，一个生产者产生的消息，可以被多个消费者进行消费，只不过多个消费者在消费消息时是竞争的关系，先得到的先消费，一旦消费完成，该消息就会出队列，
就不能被其他消费者再消费了，即“一次性消费”。就是我们熟悉的“点对点”通信了；
网站名称：实战Spring4+ActiveMQ整合实现消息队列(生产者+消费者)
网站网址：http://cqcxhl.com/article/goesjh.html