详解RabbitMQ工作模式之发布订阅模式

news2025/7/17 3:11:40

目录

发布订阅模式

概念

概念介绍

特点和优势

应用场景

注意事项

代码案例

引入依赖

常量类

编写生产者代码

编写消费者1代码

运行代码

发布订阅模式

概念

RabbitMQ的发布订阅模式（Publish/Subscribe）是一种消息传递模式，它允许消息生产者（Publisher）将消息发送到交换机（Exchange），然后交换机根据路由规则将消息广播到一个或多个队列，最后由消费者（Subscriber）从队列中接收并处理消息。

图中X表⽰交换机, 在订阅模型中，多了⼀个Exchange⻆⾊, 过程略有变化。

概念介绍

Exchange: 交换机 (X).
作⽤: ⽣产者将消息发送到Exchange, 由交换机将消息按⼀定规则路由到⼀个或多个队列中(上图中⽣产者将消息投递到队列中, 实际上这个在RabbitMQ中不会发⽣. )
RabbitMQ交换机有四种类型: fanout,direct, topic, headers, 不同类型有着不同的路由策略. AMQP协议⾥还有另外两种类型, System和⾃定义, 此处不再描述.

1. Fanout:⼴播，将消息交给所有绑定到交换机的队列(Publish/Subscribe模式)
2. Direct:定向，把消息交给符合指定routing key的队列(Routing模式)
3. Topic:通配符，把消息交给符合routing pattern(路由模式)的队列(Topics模式)
4. headers类型的交换器不依赖于路由键的匹配规则来路由消息, ⽽是根据发送的消息内容中的headers属性进⾏匹配. headers类型的交换器性能会很差,⽽且也不实⽤,基本上不会看到它的存在.

Exchange(交换机）只负责转发消息, 不具备存储消息的能⼒, 因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息就会丢失
RoutingKey: 路由键.⽣产者将消息发给交换器时, 指定的⼀个字符串, ⽤来告诉交换机应该如何处理这个消息.
Binding Key:绑定. RabbitMQ中通过Binding(绑定)将交换器与队列关联起来, 在绑定的时候⼀般会指定⼀个Binding Key, 这样RabbitMQ就知道如何正确地将消息路由到队列了.

⽐如下图: 如果在发送消息时, 设置了RoutingKey 为orange, 消息就会路由到Q1

当消息的Routing key与队列绑定的Bindingkey相匹配时，消息才会被路由到这个队列.

BindingKey其实也属于路由键中的⼀种, 官⽅解释为:the routingkey to use for the binding.
可以翻译为:在绑定的时候使⽤的路由键. ⼤多数时候,包括官⽅⽂档和RabbitMQJava API 中都把
BindingKey和RoutingKey看作RoutingKey, 为了避免混淆,可以这么理解:

1. 在使⽤绑定的时候,需要的路由键是BindingKey.
2. 在发送消息的时候,需要的路由键是RoutingKey

特点和优势

1.解耦合：生产者和消费者之间通过交换机进行解耦。生产者无需知道消息将被传递到哪些队列，消费者也无需知道消息来自哪个生产者。这种解耦合使得系统更加灵活和可扩展。
2.多播：支持多个消费者同时处理同一条消息，实现消息的多播效果。这有助于提高系统的并行处理能力和容错性。
3.灵活性：可以根据需要使用不同类型的交换机和绑定规则，以满足不同的消息传递需求。RabbitMQ提供了多种交换机类型，如直接交换机、扇形交换机、主题交换机等。

应用场景

发布订阅模式适用于需要将消息广播给多个消费者的场景，例如：

1.实时通知：如系统状态更新、订单状态变更等实时事件的通知。通过发布订阅模式，可以将这些事件广播给所有感兴趣的消费者。
2.日志记录：将应用程序的日志信息广播到多个日志处理服务进行处理和存储。这有助于实现日志的集中管理和分析。
3.事件处理：在事件驱动架构中，将事件作为消息发布到交换机，由多个消费者订阅并处理这些事件。这有助于实现事件的异步处理和分布式处理。

注意事项

1.消息持久化：为了确保消息在RabbitMQ服务器重启后不会丢失，可以将消息和队列标记为持久性。这样，即使服务器发生故障，消息仍然可以被消费者接收和处理。
2.消息确认：RabbitMQ支持消息确认机制，确保消息在成功处理后才会从队列中删除。这有助于防止消息丢失和重复处理。
3.负载均衡：在发布订阅模式中，多个消费者可以监听同一个队列或不同的队列。为了实现负载均衡，可以配置多个消费者来处理同一个队列中的消息。

代码案例
引入依赖
<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.21.0</version>
</dependency>
常量类
public class Constants {
    public static final String HOST = "47.98.109.138";
    public static final int PORT = 5672;
    public static final String USER_NAME = "study";
    public static final String PASSWORD = "study";
    public static final String VIRTUAL_HOST = "aaa";

    //发布订阅模式
    public static final String FANOUT_EXCHANGE = "fanout.exchange";
    public static final String FANOUT_QUEUE1 = "fanout.queue1";
    public static final String FANOUT_QUEUE2 = "fanout.queue2";
}
编写生产者代码
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT); //需要提前开放端口号
        connectionFactory.setUsername(Constants.USER_NAME);//账号
        connectionFactory.setPassword(Constants.PASSWORD);  //密码
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机
        Connection connection = connectionFactory.newConnection();
        //2. 开启信道
        Channel channel = connection.createChannel();
        //3. 声明交换机
        channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);
        //4. 声明队列
        channel.queueDeclare(Constants.FANOUT_QUEUE1,true,false,false,null);
        channel.queueDeclare(Constants.FANOUT_QUEUE2,true,false,false,null);
        //5. 交换机和队列绑定
        channel.queueBind(Constants.FANOUT_QUEUE1,Constants.FANOUT_EXCHANGE,"");
        channel.queueBind(Constants.FANOUT_QUEUE2,Constants.FANOUT_EXCHANGE,"");
        //6. 发布消息
        String msg = "hello fanout....";
        channel.basicPublish(Constants.FANOUT_EXCHANGE,"", null, msg.getBytes());
        System.out.println("消息发送成功");
        //7. 释放资源
        channel.close();
        connection.close();

    }
}
编写消费者1代码
import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT); //需要提前开放端口号
        connectionFactory.setUsername(Constants.USER_NAME);//账号
        connectionFactory.setPassword(Constants.PASSWORD);  //密码
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机
        Connection connection = connectionFactory.newConnection();
        //2. 开启信道
        Channel channel = connection.createChannel();
        //3. 声明队列
        channel.queueDeclare(Constants.FANOUT_QUEUE1,true,false,false,null);
        //4. 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            //从队列中收到消息, 就会执行的方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收到消息:"+ new String(body));
            }
        };
        channel.basicConsume(Constants.FANOUT_QUEUE1, true, consumer);

    }
}

编写消费者2代码

import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1. 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT); //需要提前开放端口号
        connectionFactory.setUsername(Constants.USER_NAME);//账号
        connectionFactory.setPassword(Constants.PASSWORD);  //密码
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机
        Connection connection = connectionFactory.newConnection();
        //2. 开启信道
        Channel channel = connection.createChannel();
        //3. 声明队列
        channel.queueDeclare(Constants.FANOUT_QUEUE2,true,false,false,null);
        //4. 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            //从队列中收到消息, 就会执行的方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接收到消息:"+ new String(body));
            }
        };
        channel.basicConsume(Constants.FANOUT_QUEUE2, true, consumer);

    }
}
运行代码

运行生产者代码之后，可以在管理界面看到两个队列中都有1条消息。

由此我们可以看到，交换机收到的生产者生产的一条消息被广播到了两个队列，消费者都能够分别从这两个队列中得到一条消息并消费。 

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2379817.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

微信学习之导航功能

微信学习之导航功能

先看这个功能的效果&#xff1a; 然后开始学习吧。 一、我们这里用的是vant的Grid控件&#xff0c;首先我们导入&#xff1a; { "usingComponents": {"van-search": "vant/weapp/search/index","my-swiper":"../../components…
阅读更多...
城市内涝监测预警系统守护城市安全

城市内涝监测预警系统守护城市安全

一、系统背景 城市内涝是指由于强降水或连续性降水超过城市排水能力&#xff0c;导致城市内产生积水灾害的现象。随着气候变化和城市化进程的加快&#xff0c;城市内涝现象愈发频繁和严重。传统的城市排水系统已难以满足当前的城市排水需求&#xff0c;特别是在暴雨等极端天气条…
阅读更多...
用 CodeBuddy 搭建「MiniGoal 小目标打卡器」：一次流畅的 UniApp 开发体验

用 CodeBuddy 搭建「MiniGoal 小目标打卡器」：一次流畅的 UniApp 开发体验

我正在参加CodeBuddy「首席试玩官」内容创作大赛&#xff0c;本文所使用的 CodeBuddy 免费下载链接&#xff1a;腾讯云代码助手 CodeBuddy - AI 时代的智能编程伙伴 在日常生活中&#xff0c;我们总是希望能够坚持一些小习惯&#xff0c;比如每天锻炼十分钟、读一页书、早睡十分…
阅读更多...
Web技术与Nginx网站环境部署

Web技术与Nginx网站环境部署

目录 一.web基础 1.域名和DNS &#xff08;1&#xff09;.域名的概念 &#xff08;2&#xff09;.hosts文件 &#xff08;3&#xff09;.DNS &#xff08;4&#xff09;.域名注册 2.网页与HTML &#xff08;1&#xff09;.网页简介 &#xff08;2&#xff09;.HTML &a…
阅读更多...
AI移动监测：仓储环境安全的“全天候守护者”

AI移动监测：仓储环境安全的“全天候守护者”

AI移动监测在仓储方面的应用&#xff1a;技术赋能与场景突破 一、背景&#xff1a;仓储环境的“隐形威胁”与AI破局 仓储行业长期面临设备损坏、货物损失、卫生隐患等风险。传统监控依赖人工巡检或固定摄像头&#xff0c;难以实时捕捉动态风险。例如&#xff1a; 动物入侵&a…
阅读更多...
2025年5月华为H12-821新增题库带解析

2025年5月华为H12-821新增题库带解析

IS-IS核心知识 四台路由器运行IS-IS且已经建立邻接关系&#xff0c;区域号和路由器的等级如图中标记&#xff0c;下列说法中正确的有? R2和R3都会产生ATT置位的Level-1的LSPR1没有R4产生的LSP&#xff0c;因此R1只通过缺省路由和R4通信R2和R3都会产生ATT置位的Leve1-2的LSPR2和…
阅读更多...
用 python 编写的一个图片自动分类小程序（三）

用 python 编写的一个图片自动分类小程序（三）

图片自动分类识别小程序记录 2025/5/18 0:38修改程序界面&#xff0c;增加一些功能 用 python 编写的一个图片自动识别分类小程序。 操作系统平台&#xff1a;Microsoft Windows 11 编程语言和 IDE&#xff1a;python 3.10 Visual studio code 一&#xff1a;图片自动分…
阅读更多...
初识 java

初识 java

目录 前言 一、jdk&#xff0c;JRE和JVM之间的关系 二、JVM的内存划分 前言 初步了解 jdk&#xff0c;JRE&#xff0c;JVM 之间的关系&#xff0c;JVM 的内存划分。 一、jdk&#xff0c;JRE和JVM之间的关系 jdk 是 java 开发工具集&#xff0c;包含JRE&#xff1b; JRE 是…
阅读更多...
frida 配置

frida 配置

1.环境 1.1 下载 frida-server firda-server github下载地址 这边推荐使用最新版的上一个版本 根据虚拟机自行选择版本 我使用这个版本 frida-server-16.7.17-android-x86_64 1.2 启动 frida-server-16.7.17-android-x86_64 将文件解压至虚拟机目录 使用adb命令执行 chmo…
阅读更多...
16-看门狗和RTC

16-看门狗和RTC

一、独立看门狗 1、独立看门狗概述 在由单片机构成的微型计算机系统中&#xff0c;由于单片机的工作常常会受到来自外界电磁场的干扰&#xff0c;造成程序的跑飞&#xff08;不按照正常程序进行运行&#xff0c;如程序重启&#xff0c;但是如果我们填加看门狗的技术&#xff0…
阅读更多...
【AI论文】用于评估和改进大型语言模型中指令跟踪的多维约束框架

【AI论文】用于评估和改进大型语言模型中指令跟踪的多维约束框架

摘要&#xff1a;接下来的指令评估了大型语言模型&#xff08;LLMs&#xff09;生成符合用户定义约束的输出的能力。 然而&#xff0c;现有的基准测试通常依赖于模板化的约束提示&#xff0c;缺乏现实使用的多样性&#xff0c;并限制了细粒度的性能评估。 为了填补这一空白&…
阅读更多...
Chromium 浏览器核心生命周期剖析：从 BrowserProcess 全局管理到 Browser 窗口实例

Chromium 浏览器核心生命周期剖析：从 BrowserProcess 全局管理到 Browser 窗口实例

在 Chromium 浏览器架构中&#xff0c;BrowserProcess 和 Browser 是两个核心类&#xff0c;分别管理 浏览器进程的全局状态 和 单个浏览器窗口的实例。它们的生命周期设计直接影响浏览器的稳定性和资源管理。以下是它们的详细生命周期分析&#xff1a; 1. BrowserProcess 的生…
阅读更多...
易境通海外仓系统：一件代发全场景数字化解决方案

易境通海外仓系统：一件代发全场景数字化解决方案

随着全球经济一体化和消费升级&#xff0c;一件代发业务的跨境电商市场规模持续增长。然而&#xff0c;一件代发的跨境运营也面临挑战&#xff0c;传统海外仓管理模式更因效率低下、协同困难成为业务扩张的瓶颈。 一、一件代发跨境运营痛点 1、多平台协同&#xff1a;卖家往往…
阅读更多...
【Python训练营打卡】day30 @浙大疏锦行

【Python训练营打卡】day30 @浙大疏锦行

DAY 30 模块和库的导入 知识点回顾&#xff1a; 1. 导入官方库的三种手段 2. 导入自定义库/模块的方式 3. 导入库/模块的核心逻辑&#xff1a;找到根目录&#xff08;python解释器的目录和终端的目录不一致&#xff09; 作业&#xff1a;自己新建几个不同路径文件尝试下如何…
阅读更多...
仿腾讯会议——添加音频

仿腾讯会议——添加音频

1、实现开启或关闭音频 2、 定义信号 3、实现开始暂停音频 4、实现信号槽连接 5、回收资源 6、初始化音频视频 7、 完成为每个人创建播放音频的对象 8、发送音频 使用的是对象ba&#xff0c;这样跨线程不会立刻回收&#xff0c;如果使用引用&#xff0c;跨线程会被直接回收掉&a…
阅读更多...
虚幻引擎5-Unreal Engine笔记之`GameMode`、`关卡（Level）` 和 `关卡蓝图（Level Blueprint）`的关系

虚幻引擎5-Unreal Engine笔记之`GameMode`、`关卡（Level）` 和 `关卡蓝图（Level Blueprint）`的关系

虚幻引擎5-Unreal Engine笔记之GameMode、关卡&#xff08;Level&#xff09; 和 关卡蓝图&#xff08;Level Blueprint&#xff09;的关系 code review! 参考笔记&#xff1a; 1.虚幻引擎5-Unreal Engine笔记之GameMode、关卡&#xff08;Level&#xff09; 和 关卡蓝图&…
阅读更多...
MTK zephyr平台：系统休眠流程

MTK zephyr平台：系统休眠流程

一、概述： 当内核没有需要调度的东西时，就会进入空闲状态。 CONFIG_PM=y时允许内核调用PM subsys，将空闲系统置于支持的电源状态之一。 Application负责设置唤醒事件，该事件通常是由SoC外围模块触发的中断，例如： SysTick、RTC、计数器、GPIO 并非所有外设在所有电源模式…
阅读更多...
涨薪技术|0到1学会性能测试第71课-T-SQL调优

涨薪技术|0到1学会性能测试第71课-T-SQL调优

前面的推文我们掌握了索引调优技术，今天给大家分享T-SQL调优技术。后续文章都会系统分享干货，带大家从0到1学会性能测试。 对T-SQL语句进行调校是DBA调优数据库性能的主要任务，因为不同的查询语句，即使查询出来的结果一致，其消耗的时间和系统资源也有所不同，所以如何使查…
阅读更多...
Python----目标检测（PASCAL VOC数据集）

Python----目标检测（PASCAL VOC数据集）

一、PASCAL VOC数据集 PASCAL VOC&#xff08;Visual Object Classes&#xff09;数据集是计算机视觉领域中广泛使用的一个 标准数据集&#xff0c;用于目标检测、图像分割、图像分类、动作识别等任务。该数据集由 PASCAL&#xff08;Pattern Analysis, Statistical Modelling …
阅读更多...
LabVIEW汽车CAN总线检测系统开发

LabVIEW汽车CAN总线检测系统开发

CAN&#xff08;ControllerArea Network&#xff09;总线作为汽车电子系统的核心通信协议&#xff0c;广泛应用于动力总成、车身控制、辅助驾驶等系统。基于 LabVIEW 开发 CAN 总线检测系统&#xff0c;可充分利用其图形化编程优势、丰富的硬件接口支持及强大的数据分析能力&am…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023