在上一篇《ActiveMQ入门系列二：入门代码实例(点对点模式)》中提到了ActiveMQ中的两种模式：点对点模式(PTP)和发布/订阅模式(Pub & Sub)，详细介绍了点对点模式并用代码实例进行说明，今天就介绍下发布/订阅模式。
一、理论基础
发布/订阅模式的工作示意图：
消息生产者将消息(发布)到topic中，可以同时有多个消息消费者(订阅)消费该消息。
和点对点方式不同，发布到topic的消息会被所有订阅者消费。
当生产者发布消息，不管是否有消费者，都不会保存消息。
一定要先有消息的消费者，后有消息的生产者。
二、代码实现
生产者
packagecom.sam.topic;importorg.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
public classTopicProducer {public static final String QUEUE_NAME = "topic-demo";//队列名
public void producer(String message) throwsJMSException {
ConnectionFactory factory= null;
Connection connection= null;
Session session= null;
MessageProducer producer= null;try{factory= new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");connection=factory.createConnection();connection.start();session= connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Destination destination=session.createTopic(QUEUE_NAME);
producer=session.createProducer(destination);
TextMessage textMessage=session.createTextMessage(message);producer.send(textMessage);
System.out.println("消息发送成功");
}catch(Exception ex){throwex;
}finally{
if(producer != null){
producer.close();
}if(session != null){
session.close();
}if(connection != null){
connection.close();
}
}
}public static voidmain(String[] args){
TopicProducer producer= newTopicProducer();try{
producer.producer("hello, activemq");
}catch(Exception ex){
ex.printStackTrace();
}
}
}
发布/订阅模式的生产者和点对点模式的代码主要区别就是Destination的创建方式，点对点模式是调用session.createQueue(QUEUE_NAME)，而发布/订阅模式是调用session.createTopic(QUEUE_NAME)。
消费者
packagecom.sam.topic;importorg.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;importjava.io.IOException;
public classTopicConsumer {public void consumer() throwsJMSException, IOException {
ConnectionFactory factory= null;
Connection connection= null;
Session session= null;
MessageConsumer consumer= null;try{
factory= new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616");
connection=factory.createConnection();connection.start();
session= connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination=session.createTopic(TopicProducer.QUEUE_NAME);
consumer=session.createConsumer(destination);consumer.setMessageListener(newMessageListener() {
@Overridepublic voidonMessage(Message message) {try{
TextMessage om=(TextMessage) message;
String data=om.getText();
System.out.println(data);
}catch(JMSException e) {
e.printStackTrace();
}
}
});
}catch(Exception ex){throwex;
}
}public static voidmain(String[] args){
TopicConsumer consumer= newTopicConsumer();try{
consumer.consumer();
}catch(Exception ex){
ex.printStackTrace();
}
}
}
消费者在点对点监听消费的基础上进行变化，主要区别有两个：1.同生产者一样，也是Destination的创建方式不同； 2.消息无需手动确认，直接采用自动确认机制
代码写完了，接下来进行测试，由于subscribe可以有多个，而且每个都可以消费到相同的消息，因此我们消费者启动两个。
先执行生产者
在控制台页面的Topics下出现了我定义的topic并且有1条消息发送成功且未消费
然后执行两个消费者，两个消费者都没有消费到任何消息
并且，控制台页面只是多了2个消费者，已经消费的消息还是0
为什么呢？还记得前面的理论基础说的吗？就是这个原因
继续，我们在两个消费者启动好的前提下，再执行生产者， 这个时候会发现两个消费者都消费了该消息
再看下控制台页面
已消费消息这里是2，这个2并不是说之前发的两个消息都消费了，而是说第二个消息消费了2次， 1 * 2 = 2
不信的话，可以再执行一遍生产者，这个时候就是4，而不是3
累计发送过3条消息，消息消费了4次，这里的4就是后面两条分别被消费了2次， 2 * 2 = 4
三、两种模式比较
好，到这里，发布/订阅模式就介绍完了。
如果有收获，就点个赞呗