ActiveMQ入门系列三：发布/订阅模式

在上一篇《activemq入门系列二：入门代码实例（点对点模式）》中提到了activemq中的两种模式：点对点模式（ptp）和发布/订阅模式（pub & sub），详细介绍了点对点模式并用代码实例进行说明，今天就介绍下发布/订阅模式。

一、理论基础

发布/订阅模式的工作示意图：

• 消息生产者将消息（发布）到topic中，可以同时有多个消息消费者（订阅）消费该消息。
• 和点对点方式不同，发布到topic的消息会被所有订阅者消费。
• 当生产者发布消息，不管是否有消费者，都不会保存消息。
• 一定要先有消息的消费者，后有消息的生产者。

二、代码实现

1. 生产者

   package com.sam.topic;
   
   import org.apache.activemq.activemqconnectionfactory;
   
   import javax.jms.*;
   
   /**
    * @author java开发老菜鸟
    *
    */
   public class topicproducer {
   
       public static final  string queue_name = "topic-demo";//队列名
   
       public void producer(string message) throws jmsexception {
           connectionfactory factory = null;
           connection connection = null;
           session session = null;
           messageproducer producer = null;
           try {
               /**
                * 1.创建连接工厂
                * 创建工厂，构造方法有三个参数：分别是用户名、密码、连接地址
                * 无参构造：有默认的连接地址，localhost
                * 一个参数：无验证模式，无用户的认证
                * 三个参数：有认证和连接地址
                */
               factory = new activemqconnectionfactory("admin","admin","tcp://localhost:61616");
               /**
                * 2.创建连接
                * 无参数
                * 有参数：用户名、密码
                */
               connection = factory.createconnection();
               /**
                * 3.启动连接
                * 生产者可以不启动，因为在发送消息的时候回进行检查
                * 如果未启动连接，会自动启动
                * 如果有特殊配置，需要配置完成后再启动连接
                */
               connection.start();
               /**
                * 4.用连接创建会话
                * 有两个参数：是否需要事务、消息确认机制
                * 如果支持事务，对于生产者来说第二个参数就无效了，建议传入session.session_transacted
                * 如果不支持事务，第二个参数必须传递且有效
                *
                * auto_acknowledge：自动确认，消息处理后自动确认（商业开发不推荐）
                * client_acknowledge：客户端手动确认，消费者处理后必须手动确认
                * dups_ok_acknowledge：有副本的客户端手动确认，消息可以多次处理（不建议）
                */
               session = connection.createsession(false, session.auto_acknowledge);
               /**
                * 5.用会话创建目的地（主题）、生产者、消息
                * 队列名是队列的唯一标记
                * 创建生产者的时候可以不指定目的地，可以在发送的时候指定
                */
               destination destination = session.createtopic(queue_name);
               producer = session.createproducer(destination);
               textmessage textmessage = session.createtextmessage(message);
               /**
                * 6.生产者发送消息到目的地
                */
               producer.send(textmessage);
               system.out.println("消息发送成功");
           } catch(exception ex){
               throw ex;
           } finally {
               /**
                * 7.释放资源
                */
               if(producer != null){
                   producer.close();
               }
   
               if(session != null){
                   session.close();
               }
   
               if(connection != null){
                   connection.close();
               }
           }
       }
   
       public static void main(string[] args){
           topicproducer producer = new topicproducer();
           try{
               producer.producer("hello, activemq");
           } catch (exception ex){
               ex.printstacktrace();
           }
       }
   }

   发布/订阅模式的生产者和点对点模式的代码主要区别就是destination的创建方式，点对点模式是调用session.createqueue(queue_name)，而发布/订阅模式是调用session.createtopic(queue_name)。

2. 消费者

   package com.sam.topic;
   
   import org.apache.activemq.activemqconnectionfactory;
   
   import javax.jms.*;
   import java.io.ioexception;
   
   /**
    * @author java开发老菜鸟
    *
    * 观察者消费--监听消费
    */
   public class topicconsumer {
   
       public void consumer() throws jmsexception, 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);
               /**
                * 注册监听器，队列中的消息变化会自动触发监听器，接收并自动处理消息
                *
                * 监听器一旦注册，永久有效，一直到程序关闭
                * 监听器可以注册多个，相当于集群
                * activemq自动轮询多个监听器，实现并行处理
                */
               consumer.setmessagelistener(new messagelistener() {
                   @override
                   public void onmessage(message message) {
   
                       try {
                           textmessage om = (textmessage) message;
                           string data = om.gettext();
                           system.out.println(data);
                       } catch (jmsexception e) {
                           e.printstacktrace();
                       }
                   }
               });
           } catch(exception ex){
               throw ex;
           }
       }
   
       public static void main(string[] args){
           topicconsumer consumer = new topicconsumer();
           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

 

三、两种模式比较

 

好，到这里，发布/订阅模式就介绍完了。

如果有收获，就点个赞呗