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

Posted 2021-12-29 sam-uncle

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ActiveMQ入门系列三：发布/订阅模式相关的知识，希望对你有一定的参考价值。

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

一、理论基础

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

技术图片

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

二、代码实现

生产者

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)。

消费者

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();