3，ActiveMQ-入门（基于JMS发布订阅模型）

一、Pub/Sub-发布/订阅消息传递模型

在发布/订阅消息模型中，发布者发布一个消息，该消息通过topic传递给所有的客户端。在这种模型中，发布者和订阅者彼此不知道对方，是匿名的且可以动态发布和订阅topic。topic主要用于保存和传递消息，且会一直保存消息直到消息被传递给客户端。

发布订阅模型就像订阅报纸。我们可以选择一份或者多份报纸，比如：北京日报、人民日报。这些报纸就相当于发布订阅模型中的topic。如果有很多人订阅了相同的报纸，那我们就在同一个topic中注册，对于报纸发行方，它就和所有的订阅者形成了一对多的关系。如下：

二，Pub/Sub特点

1，每个消息可以有多个消费者。

2，发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个或多个订阅者之后，才能消费发布者的消息，而且为了消费消息，订阅者必须保持运行的状态。

3，为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。

三，发布订阅模型的实现

3.1，发布者

/*** * @类名称：ActiveMQpubsubProducter* @类描述：发布者-发布订阅模型*/
public class ActiveMQpubsubProducter {//会话对象private static Session session = null;public static void sendMessages(String name){try {session = ActiveMQLinkUtil.initConnection(true, Session.AUTO_ACKNOWLEDGE);if(session != null){//创建一个主题Topic messageTopic = session.createTopic(name);//创建消息发布者  MessageProducer messageProducer = session.createProducer(messageTopic);for (int i = 0; i < 5 ; i++) {TextMessage  t = session.createTextMessage("ActiveMQ发布消息：" + i);messageProducer.send(t);}session.commit();}} catch (Exception e) {e.printStackTrace();}}public static void main(String[] args) {ActiveMQpubsubProducter.sendMessages("topicDemo");}
}

3.2，订阅者

----------------------------订阅者1----------------------------

/*** * @类名称：ActiveMQpubsubComsumer1* @类描述：订阅者1-发布订阅模型*/
public class ActiveMQpubsubComsumer1 {//会话对象private static Session session = null;public static void getMessages(String name){try {session = ActiveMQLinkUtil.initConnection(false, Session.AUTO_ACKNOWLEDGE);if(session != null){Topic topic = session.createTopic(name);//创建消费者  MessageConsumer messageConsumer = session.createConsumer(topic);//注册消息监听  messageConsumer.setMessageListener(new MyListener1());}} catch (Exception e) {e.printStackTrace();}}public static void main(String[] args) {ActiveMQpubsubComsumer1.getMessages("topicDemo");}
}

----------------------------订阅者2----------------------------

/*** * @类名称：ActiveMQpubsubComsumer2* @类描述：订阅者2-发布订阅模型*/
public class ActiveMQpubsubComsumer2 {//会话对象private static Session session = null;public static void getMessages(String name){try {session = ActiveMQLinkUtil.initConnection(false, Session.AUTO_ACKNOWLEDGE);if(session != null){Topic topic = session.createTopic(name);//创建消费者  MessageConsumer messageConsumer = session.createConsumer(topic);//注册消息监听  messageConsumer.setMessageListener(new MyListener2());}} catch (Exception e) {e.printStackTrace();}}public static void main(String[] args) {ActiveMQpubsubComsumer2.getMessages("topicDemo");}
}

3.3，监听类

----------------------------订阅者1-监听类----------------------------

/*** * @类名称：MyListener1* @类描述：订阅者1-监听类*/
public class MyListener1 implements MessageListener {@Overridepublic void onMessage(Message message) {try {System.out.println("订阅者一，收到的消息：" + ((TextMessage)message).getText());//签收
            message.acknowledge();} catch (JMSException e) {e.printStackTrace();}}
}

----------------------------订阅者2-监听类----------------------------

/*** * @类名称：MyListener2* @类描述：订阅者2-监听类*/
public class MyListener2 implements MessageListener {@Overridepublic void onMessage(Message message) {try {System.out.println("订阅者二，收到的消息：" + ((TextMessage)message).getText());//签收
            message.acknowledge();} catch (JMSException e) {e.printStackTrace();}}
}

四，测试

在点对点的要先启动生产者，生产者要生产消息。而发布订阅模型，要先启动订阅者，订阅者先订阅topic，再发布消息。

第一步：运行ActiveMQpubsubComsumer1，ActiveMQpubsubComsumer2

第二步：运行ActiveMQpubsubProducter

五，结果

ActiveMQpubsubComsumer1：

ActiveMQpubsubComsumer2：

ActiveMQpubsubProducter：

ActiveMQ控制台截图：

Number Of Consumers	在该队列上还有多少消费者在等待接受消息。
Messages Dequeued	消费了多少条消息，记做C。
Messages Enqueued	生产了多少条消息，记做P。

发布者发布了5条数据，但是出队的有10条，因为有两个订阅者。

六，总结

发布者向一个特定的消息主题发布消息，0或者多个订阅者可能接收到来自特定消息主题的消息感兴趣。其中发布者和订阅者不知道对方的存在。

转载于:https://www.cnblogs.com/Zender/p/9101003.html