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

在上一篇《ActiveMQ入门系列二:入门代码实例(点对点模式)》中提到了ActiveMQ中的两种模式:点对点模式(PTP)和发布/订阅模式(Pub&Sub),详细介绍了点对点模式并用代码

大家好,又见面了,我是全栈君。

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

一、理论基础

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

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

  • 消息生产者将消息(发布)到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可以有多个,而且每个都可以消费到相同的消息,因此我们消费者启动两个。

先执行生产者

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

在控制台页面的Topics下出现了我定义的topic并且有1条消息发送成功且未消费

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

 

然后执行两个消费者,两个消费者都没有消费到任何消息

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

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

并且,控制台页面只是多了2个消费者,已经消费的消息还是0

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

 

 为什么呢?还记得前面的理论基础说的吗?就是这个原因

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

继续,我们在两个消费者启动好的前提下,再执行生产者, 这个时候会发现两个消费者都消费了该消息

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

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

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

再看下控制台页面

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

已消费消息这里是2,这个2并不是说之前发的两个消息都消费了,而是说第二个消息消费了2次, 1 * 2 = 2

不信的话,可以再执行一遍生产者,这个时候就是4,而不是3

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

累计发送过3条消息,消息消费了4次,这里的4就是后面两条分别被消费了2次, 2 * 2 = 4

 

三、两种模式比较

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

 

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

如果有收获,就点个赞呗

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/120886.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)
blank

相关推荐

  • 优化算法——人工蜂群算法(ABC)

    优化算法——人工蜂群算法(ABC)一、人工蜂群算法的介绍人工蜂群算法(ArtificialBeeColony,ABC)是由Karaboga于2005年提出的一种新颖的基于群智能的全局优化算法,其直观背景来源于蜂群的采蜜行为,蜜蜂根据各自的分工进行不同的活动,并实现蜂群信息的共享和交流,从而找到问题的最优解。人工蜂群算法属于群智能算法的一种。二、人工蜂群算法的原理1、原理标准的ABC算法通过模拟

  • 第二章_session管理

    第二章_session管理

  • 简单区分单射、满射与双射

    简单区分单射、满射与双射我的机器学习教程「美团」算法工程师带你入门机器学习已经开始更新了,欢迎大家订阅~任何关于算法、编程、AI行业知识或博客内容的问题,可以随时扫码关注公众号「图灵的猫」,加入”学习小组“,沙雕博主在线答疑~此外,公众号内还有更多AI、算法、编程和大数据知识分享,以及免费的SSR节点和学习资料。其他平台(知乎/B站)也是同名「图灵的猫」,不要迷路哦~利用泛…

  • input = file 实现上传文件

    input = file 实现上传文件使用input=file实现文件上传功能,如果点击提交按钮,提交表单并上传选中的文件:<FormItemlabel=”上传简历”><divclassName=”section-pushInChannel-fileInput”><inputid=”file”onChange={this.handleFileChange}t…

  • 基于FPGA的CAN接口开发

    基于FPGA的CAN接口开发基于Xilinx的A7系列FPGA的CAN总线协议开发一、CAN总线协议介绍CAN是ControllerAreaNetwork的缩写(以下称为CAN),是ISO国际标准化的串行通信协议。可以用来满足“多总线通信时,线束的数量过多”、“通过多个LAN,进行大量数据的高速通信”的需要。它的出现为分布式控制系统实现各节点之间实时、可靠的数据通信提供了强有力的技术支持。CAN控制器根据两根线上的电位差来判断总线电平。总线电平分为显性电平和隐性电平,二者必居其一。发送方通过使总线电平发生变化

  • 函数指针赋值

    函数指针赋值在远程注入的时候特别需要给函数指针赋值。 有以下2种方法,第一定义方法。主要用于给大量同参数的函数注入。 #include”stdafx.h”#include”windows.h”typedefint(_stdcall*p_MessageBoxA)(HWNDhWnd,LPCSTRlpText,LPCSTRlpCaption,UINTuType)

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号