1、ActiveMQ中消息的管理机制：使用ActiveMQ的目的必然是处理消息，大体步骤如下：
1）通过ConnectionFactory连接到ActiveMQ服务器
2）通过ConnectionFactory创建Connection
3）通过Connection获取Session
4）通过Session创建消息的目的地，即队列(Queue)或主题(Topic)
5）通过Session创建消息生产者，生产的目的地。即4）中创建的Queue或Topic
6）通过Session创建消息
7）通过消息生产者将消息发送至消息的目的地
8）消费者在相同的消息目的地消费消息
注：至于消息的目的地应该是队列(Queue)还是主题(Topic)，需要看该消息是只能被消费一次、还是需要被消费多次，也就是说Queue中的消息适合点对点的模式，而Topic中的消息则适合发布/订阅模式。Queue和Topic都是Destination接口的实现
2、消费目的地的两种模式：
1）点对点(P2P)模型
点对点模型，采用的是队列(Queue)作为消息载体。在该模式中，一条消息只能被一个消费者消费，没有被消费的，只能留在队列中，等待被消费，或者超时。
2）发布/订阅(Pub/Sub)模型
Topic模式适用于发布/订阅，生产者将消息发布到Topic中，每个消息都可能有多个消费者，属于1:N的关系，生产者和消费者之间有时间上的相关性，订阅某一个主题的消费者只能消费自它订阅之后生产者发布的消息，如果生产者在发布消息时没有订阅者，那发布的消息就是一条垃圾消息，因此生产者在发布消息时应该检查当前是否有订阅者。
JMS规范还允许客户创建持久订阅，这在一定程度上放松了时间上的相关性，持久订阅允许消费者消费它在未处于激活状态时生产者发送到主题的消息。
发布/订阅模型采用的是主题(Topic)作为消息通讯载体。该模式类似微信公众号的模式。发布者发布一条信息，然后将该信息传递给所有的订阅者。注意：订阅者想要接收到该信息，必须在该信息发布之前订阅。
3、Message Groups(消息分组)
逻辑上，可以看成是一种并发的Exclusive Consumer(排它消费)。JMS消息属性JMXGroupID被用来区分Message Group。
消息分组的特性：
a.保证所有具有相同JMSGroupID的消息会被分发到相同的消费者(只要这个Consumer保持Active);
b.Message Groups也是一种负载均衡的机制;
在一个消息被分发到Consumer前，Broker会检查消息的JMSGroupID属性。如果存在，那么broker会检查是否有某个Consumer拥有这个Message Group。如果没有，那么broker会选择一个Consumer，并将它关联到这个Message Group。此后，这个Consumer会接收这个Message Group的所有消息，直到Consumer被关闭。
从4.1版本开始，ActiveMQ支持一个布尔字段JMSXGroupFirstForConsumer。当某个message group的第一个消息被发送到consumer的时候，这个字段被设置。如果客户使用failover transport连接到broker。在由于网络问题等造成客户重新连接到broker的时候，相同message group的消息可能会被分发到不同与之前的consumer，因此JMSXGroupFirstForConsumer字段也会被重新设置。
4、点对点模式和发布/订阅(Pub/Sub)模式下生产消息和消费消息示例：
4.1 依赖引入：
<!--activemq依赖引入-->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
##activemq service info
activemq.address=127.0.0.1
activemq.port=61616
activemq.username=admin
activemq.password=admin
4.2 代码示列
1）生产和消费请求：ActiveMqQueueController
package com.zj.weblearn.controller.activemq; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody; import java.util.HashMap; import java.util.Map; /* * @Copyright (C), 2002-2020, * @ClassName: ActiveMqApp * @Author: * @Date: 2020/9/24 15:49 * @Description: * @History: * @Version:1.0 */ @Controller() @RequestMapping("/activemq/queue") public class ActiveMqQueueController { @Autowired com.zj.weblearn.serviceImpl.activemq.QueueMessageServiceImpl queueMessageServiceImpl; //点对点模式：生产者生产消息 http://localhost:8080/activemq/queue/produceMess.do @RequestMapping("/produceMess") @ResponseBody public Map producerSendMessage(){ Map resultMap=new HashMap(); resultMap.put("isSuccess",false); Map<String,String> message=new HashMap(); for(int i=0;i<10;i++){ message.put(String.valueOf(i),"this is "+i+" message"); } message.put(String.valueOf(10),""); try { //向队列中发送消息 queueMessageServiceImpl.p2pProducerSendMess(message); resultMap.put("isSuccess",true); }catch (Exception e){ resultMap.put("errorMessage","生产消息异常"); } return resultMap; } //消费者消费消息 http://localhost:8080/activemq/queue/consumeMess.do @RequestMapping("/consumeMess") @ResponseBody public Map consumeMessage(){ Map resultMap=new HashMap(); resultMap.put("isSuccess",false); try {//消费队列中消息的三种方法： //queueMessageServiceImpl.p2pconsumerConsumeMessMethod1ByReceive();//消费方法1 //queueMessageServiceImpl.p2pconsumerConsumeMessMethod2BySetReceiveTime();//消费方法2 queueMessageServiceImpl.p2pconsumerConsumeMessMethod3ByMessageListener();//消费方法3 resultMap.put("isSuccess",true); }catch (Exception e){ resultMap.put("errorMessage","消息消费异常"+e); } return resultMap; } //一边生产一边消费 http://localhost:8080/activemq/queue/sideProductSideConsume.do @RequestMapping("/sideProductSideConsume") @ResponseBody public Map sideProductSideConsumeMessage(){ Map resultMap=new HashMap(); resultMap.put("isSuccess",false); Map<String,String> message=new HashMap(); try {// queueMessageServiceImpl.sideProductSideConsumeMessage(message);//消费方法3 resultMap.put("isSuccess",true); }catch (Exception e){ resultMap.put("errorMessage","消息消费异常"+e); } return resultMap; } //消息分组 http://localhost:8080/activemq/queue/messageGroupBy.do @RequestMapping("/messageGroupBy") @ResponseBody public Map messageGroupBy(){ Map resultMap=new HashMap(); resultMap.put("isSuccess",false); Map<String,String> message=new HashMap(); try {// queueMessageServiceImpl.messageConsumedByGroup(message);//消费方法3 resultMap.put("isSuccess",true); }catch (Exception e){ resultMap.put("errorMessage","消息消费异常"+e); } return resultMap; } }
2）创建连接工具方法：ActiveMqConnectionUtils
import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.JMSException; /* * @Copyright (C), 2002-2020, * @ClassName: RabbitMqConnectionUtils * @Author: * @Date: 2020/9/23 9:42 * @Description: * @History: * @Version:1.0 */ public class ActiveMqConnectionUtils { public static javax.jms.Connection getConnection() { javax.jms.Connection connection = null; try { javax.jms.ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ReadPropertiesUtils1.getValue("activemq.username"), ReadPropertiesUtils1.getValue("activemq.password"), "tcp://127.0.0.1:61616"); connection = connectionFactory.createConnection(); connection.start(); } catch (JMSException e) { e.printStackTrace(); } return connection; } }
3）点对点模式：生产和消费消息方法生产和消费消息方法(包含：receive()阻塞消费、receive(5000)消费时设置超时时间、consumer.setMessageListener(new MessageListener(){})消费时设置监听是非阻塞的;)
import com.zj.weblearn.utils.ActiveMqConnectionUtils; import org.apache.activemq.broker.jmx.CompositeDataConstants; import org.apache.commons.lang3.StringUtils; import org.springframework.stereotype.Service; import javax.jms.*; import java.io.UnsupportedEncodingException; import java.util.Map; /* * @Copyright (C), 2002-2020, * @ClassName: QueueMessageServiceImpl * @Author: * @Date: 2020/9/25 10:07 * @Description: * @History: * @Version:1.0 */ @Service public class QueueMessageServiceImpl { /* * 实现步骤 * 1.建立ConnectionFactory工厂对象，需要填入用户名、密码、连接地址（一般使用默认，如果没有修改的话） * 2.通过ConnectionFactory对象创建一个Connection连接，并且调用Connection的start方法开启连接，Connection方法默认是关闭的 * 3.通过Connection对象创建Session会话（上下文环境对象），用于接收消息，参数1是是否启用事物，参数2是签收模式，一般设置为自动签收 * 4.通过Session对象创建Destination对象，指的是一个客户端用来制定生产消息目标和消费消息来源的对象。在PTP的模式中，Destination被称作队列，在Pub/Sub模式中，Destination被称作主题（Topic） * 5.通过Session对象创建消息的发送和接收对象（生产者和消费者） * 6.通过MessageProducer的setDeliverMode方法为其设置持久化或者非持久化特性 * 7.使用JMS规范的TextMessage形式创建数据（通过Session对象），并用MessageProducer的send方法发送数据。客户端同理。记得关闭 */ //点对点模式，生产者生产消息 public void p2pProducerSendMess(Map<String, String> message) { Connection connection = ActiveMqConnectionUtils.getConnection(); try { Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("OrderCancleQueue"); MessageProducer producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); for (Map.Entry<String, String> entry : message.entrySet()) { TextMessage textMessage = session.createTextMessage(); MapMessage mapMessage=session.createMapMessage(); textMessage.setText(entry.getValue().toString()); producer.send(textMessage); } //此处有一个奇怪的现象，生产者还未把自身的10个消息全部发送完成，消费者就开始消费了，并且消费消息的顺序没有保证 System.out.println("p2pProducerSendMess>>>send message end"); if (connection != null) { connection.close(); } } catch (JMSException e) { e.printStackTrace(); } } //点对点模式，消费者消费消息 总共有3种，方法1：通过receive()方法实现 public void p2pconsumerConsumeMessMethod1ByReceive() { Connection connection = ActiveMqConnectionUtils.getConnection(); try { Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("OrderCancleQueue"); MessageConsumer consumer = session.createConsumer(destination); while (true) { /* consumer.receive()是一个阻塞方法，在接收不到队列中的消息时就一直阻塞直到接收到消息为止，因此while循环是不可能被中断的，下面的资源也不会得到释放，就会一直和ActiveMQ保持连接。 */ TextMessage textMessage = (TextMessage) consumer.receive(); String messageInfo=textMessage.getText(); if(StringUtils.isNotEmpty(messageInfo)){ System.out.println("p2pconsumerConsumeMessMethod1>>>consumer>>consume>>message>>"+new String (textMessage.getText().getBytes("iso8859-1"),"UTF-8")); }else{ System.out.println("p2pconsumerConsumeMessMethod1>>>consumer>>consume>>message>> end"); break; } } } catch (JMSException | UnsupportedEncodingException e) { e.printStackTrace(); } } //点对点模式，消费者消费消息 总共有3种，方法2：通过receive(Long var) 设置超时时间实现 public void p2pconsumerConsumeMessMethod2BySetReceiveTime() { Connection connection = ActiveMqConnectionUtils.getConnection(); try { Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("OrderCancleQueue"); MessageConsumer consumer = session.createConsumer(destination); while (true) { /* consumer.receive(4000l) */ TextMessage textMessage = (TextMessage) consumer.receive(5000);//5s String messageInfo=textMessage.getText(); if(StringUtils.isNotEmpty(messageInfo)){ System.out.println("p2pconsumerConsumeMessMethod1>>>consumer>>consume>>message>>"+new String (textMessage.getText().getBytes("iso8859-1"),"UTF-8")); }else{ System.out.println("p2pconsumerConsumeMessMethod1>>>consumer>>consume>>message>> end"); break; } } } catch (JMSException | UnsupportedEncodingException e) { e.printStackTrace(); } } //点对点模式，消费者消费消息 总共有3种，方法3：采用消息监听器：messageListener /* * 1）使用消息监听(即MessageListener)的方式也是非阻塞的，在监听不到队列中的消息时不会一直等待 * 2）使用消息监听的方式在监听到有消息时会开启一个新的线程处理消息，因此如果我们在主线程中释放了资源就可能会看不到onMessage()方法中的打印结果 * 3）使用消息监听的方式一旦我们释放了连接等资源，在生产者再次向监听的队列中发送消息时就监听不到新发送的消息了，这就失去了监听的意义，那么问题来了：使用消息监听的方式到底该不该释放连接等资源呢？释放了就会监听不到新的消息，不就失去了监听的意义了吗？不释放又会占用系统资源，应该怎么办呢？ * * 答案是不释放，因为MQ是用在分布式系统内部的，用于各个系统之间的消息通信(所有系统的总量其实并不会很多)，对于消费者端而言基本上一个队列或者一个Topic的连接数最多是分布式子系统的个数(在集群环境下最多是子系统的个数的n倍，n的值并不会很大，也就是每个子系统都要消费所有队列和订阅所有Topic的消息时的所需的连接数是最多的)，因此连接数并不会很多(估算的话最多是：队列的个数 * 系统的总个数 * 集群的倍数 + Topic的个数 * 系统的总个数 * 集群的倍数)，所以没必要考虑负载和系统资源占用的问题。虽然是这样，但我们并不能随便使用MQ，不能仅仅为了一个异步处理就开启并占用一个MQ的连接，在异步处理的时候可以采用开启线程的方式 * * 4）ActiveMQ队列中的消息在消费时采用的是轮询负载的机制，即每个消费者轮流消费消息，当然负载机制是可以配置的 * * */ public void p2pconsumerConsumeMessMethod3ByMessageListener() { Connection connection = ActiveMqConnectionUtils.getConnection(); try { //【注：第一个参数表示是否开启事务（如果采用事务方式，在发送完消息后需要关闭session，否则消息不能发送出去）；第二个参数为应答模式，我们使用生产一次，消费一次，自动应答这种模式；】 Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("OrderCancleQueue"); MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { TextMessage msg = (TextMessage) message; String text = null; try { text = msg.getText(); } catch (JMSException e) { e.printStackTrace(); } System.out.println(">>son thread：" + Thread.currentThread().getName()); System.out.println(text); } }); } catch (JMSException e) { e.printStackTrace(); } } public void sideProductSideConsumeMessage(Map<String, String> message) { for(int i=0;i<10;i++){ message.put(String.valueOf(i),"this is "+i+" message"); } Connection connection = ActiveMqConnectionUtils.getConnection(); try { Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("OrderCancleQueue"); MessageProducer producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); for (Map.Entry<String, String> entry : message.entrySet()) { TextMessage textMessage = session.createTextMessage(); textMessage.setText(entry.getValue().toString()); //发送消息 producer.send(textMessage); } MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { TextMessage msg = (TextMessage) message; String text = null; try { text = msg.getText(); } catch (JMSException e) { e.printStackTrace(); } System.out.println(">>son thread>>" + Thread.currentThread().getName()+"consume message"+text); } }); } catch (JMSException e) { e.printStackTrace(); } } public void messageConsumedByGroup(Map<String, String> message) { for(int i=0;i<10;i++){ message.put(String.valueOf(i),"this is "+i+" message"); } Connection connection = ActiveMqConnectionUtils.getConnection(); try { Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("OrderCancleQueue"); MessageProducer producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); int i=0; for (Map.Entry<String, String> entry : message.entrySet()) { i++; TextMessage textMessage = session.createTextMessage(); textMessage.setStringProperty(CompositeDataConstants.JMSXGROUP_ID, i % 2 + ""); textMessage.setText(entry.getValue().toString()); //发送消息 producer.send(textMessage); //同时创建多个消费者，以便进行不同组中信息的消费 MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { TextMessage msg = (TextMessage) message; String text = null; try { text = msg.getText(); System.out.println(">>son thread>>" + Thread.currentThread().getName()+" consume message("+text+") group name>>"+message.getStringProperty(CompositeDataConstants.JMSXGROUP_ID)); } catch (JMSException e) { e.printStackTrace(); } } }); } } catch (JMSException e) { e.printStackTrace(); } } }
4）发布/订阅(Pub/Sub)模型：生产和消费消息方法
import com.zj.weblearn.utils.ActiveMqConnectionUtils; import org.springframework.stereotype.Service; import javax.jms.*; import java.util.Map; /* * @Copyright (C), 2002-2020, * @ClassName: TopicMessageServiceImpl * @Author: * @Date: 2020/9/24 16:11 * @Description: * @History: * @Version:1.0 */ @Service public class TopicMessageServiceImpl { //发布/订阅(Pub/Sub)模型，发布者发布信息： public void pubSubProducerPubMess(Map<String,String> message) { Connection connection = ActiveMqConnectionUtils.getConnection(); try { Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 5、使用Session对象创建一个Destination对象。两种形式queue、topic，现在应该使用topic Topic topic = session.createTopic("test-topic"); // 6、使用Session对象创建一个Producer对象。 MessageProducer producer = session.createProducer(topic); // 7、创建一个Message对象，可以使用TextMessage。 for(Map.Entry entry:message.entrySet()){ TextMessage textMessage = session.createTextMessage("send message " + entry.getValue()); // 8、发送消息 producer.send(textMessage); } // 9、关闭资源 producer.close(); session.close(); connection.close(); } catch (JMSException e) { e.printStackTrace(); } } //发布/订阅(Pub/Sub)模型，订阅者订阅信息 public void pubSubConsumerSubMess() { Connection connection = ActiveMqConnectionUtils.getConnection(); try { Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建一个Destination对象。topic对象 Topic topic = session.createTopic("test-topic"); // 使用Session对象创建一个消费者对象。 MessageConsumer consumer = session.createConsumer(topic); System.out.println("topic consumer start sub message"); // 接收消息 consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { // 打印结果 TextMessage textMessage = (TextMessage) message; String text; try { text = textMessage.getText(); System.out.println("this is recived messages>>" + text); } catch (JMSException e) { e.printStackTrace(); } } }); // 关闭资源 //consumer.close(); // session.close(); // connection.close(); } catch (JMSException e) { e.printStackTrace(); } } }
5、控制台中的相关描述：http://localhost:8161/admin/
Number Of Consumers 消费者 这个是消费者端的消费者数量 。
Number Of Pending Messages 等待消费的消息 这个是当前未出队列的数量。可以理解为总接收数-总出队列数
Messages Enqueued 进入队列的消息 进入队列的总数量,包括出队列的。 这个数量只增不减
Messages Dequeued 出了队列的消息 可以理解为是消费这消费掉的数量