当前位置： 首页 > news >正文

【RabbitMQ】工作模式

news 来源：原创 2024/9/21 22:49:28

工作模式概述

简单模式

 简单模式中只存在一个生产者，只存在一个消费者。生产者生产消息，消费者消费消息。消息只能被消费一次，也称为点对点模式。

简单模式适合在消息只能被单个消费者处理的场景下存在。

工作队列模式（Work Queue）

工作队列模式中存在一个消费者，多个生产者。生产者生产消息，消息队列将生产的消息分发给不同的消费者，每个消费者接收到不同的消息后开始进行消费。简单来说，工作模式下，消息不会被重复消费，不同的消费者消费的是不同的消息。

工作模式适合在集群环境中做异步处理。

发布订阅模式

交换机（Exchange）

作用：生产者将消息发送到Broker中，会先经过交换机，由交换机将消息按照一定规则路由到一个或者多个消息队列中（在简单模式和工作队列模式下，由生产者直接将消息投递到队列中，这种情况在RabbitMQ中根本不会出现）。

RabbitMQ交换机有四种类型：fanout、direct、topic、headers。不同类型有着不同的路由策略。AMQP协议其实是有六种交换机类型的（除了上述四种，还有system和自定义），只不过RabbitMQ只使用了其四种而已。

1. Fanout：广播，交换机将从生产者中获取的消息交给与之绑定的全部队列（对应工作模式中的发布订阅模式）。

2. Direct：定向，交换机将从生产者中获取的消息交给与之绑定的符合RoutingKey的队列（对应工作模式中的路由模式）。具体RoutingKey是啥，后面会讲到。

3. Topic：通配符，交换机将从生产者中获取的消息交给与之绑定的符合RoutingKey的队列（对应工作模式中的通配符模式）。定向和通配符中的RoutingKey是略有不同的，具体到工作模式的路由模式和通配符模式就会明白。

4. headers：此类交换器并不依赖于RoutingKey的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。headers类型的交换器性能会很差，而且也不实用，基本看不到它的存在，了解即可。

Exchange只负责转化消息，并不负责存储消息。因此如果没有任何交换机和队列绑定，或者发送的消息没有符合路由规则的队列，消息就会丢失。

RoutingKey，路由键。生产者发送消息给Broker时，指定的一个字符串，用来告诉交换机应该如何处理这个消息。

BindingKey，绑定键。在声明交换机和队列之后，使用一个BindingKey将交换机和队列绑定起来。这样，当生产者将消息发送给Broker之后，交换机接收到消息就能根据消息中的RoutingKey和BindingKey进行对比，从而知道将消息路由到某个或者某几个队列中了。

本质上，BindingKey属于RoutingKey的一种。换句话说，两者的作用并没有什么差别。不同的是，路由键是生产者向Broker发送消息时使用的，绑定键则是交换机和队列绑定时进行绑定，然后再交换机给队列发送消息时使用。当生产者将一个绑定了RoutingKey的消息发送给交换机之后，交换机根据存在的BindingKey来将消息路由给队列。

发布订阅模式存在一个生产者，多个消费者。生产者生产消息，交换机将消息复制多份，每个队列都会接收到相同的消息，每个消费者接收到消息之后开始进行消费。简单来说，消费者发送的消息，所有与之关联的队列都会收到相同的消息。

发布订阅模式适合消费需要被多个消费者同时接收的场景，例如实时播报或者广播消息。

路由模式

 路由模式是发布订阅模式的变种，在发布订阅模式的基础上，增加了路由键。也就是说，消息到达交换机之后，不再是分发给所有关联的队列，而是根据绑定的路由规则来进行分发消息。

路由模式适合需要根据特定规则分发消息的场景。例如，系统日志打印，将不同级别的日志发送到不同的队列，最终输出到不同的文件。

通配符模式

通配符模式又是路由模式的变种，在路由模式的基础上，增加了通配符的功能，使消息分发更加灵活。

总的来说，发布订阅模式是消息到达交换机之后，交换机无条件的将所有消息转发给队列。路由模式是消息到达交换机之后，交换机根据RoutingKey的规则，将数据筛选之后分发给不同的队列。通配符模式也是消息到达交换机之后，交换机根据RoutingKey的规则，将数据筛选之后分发给不同的队列，只不过该RoutingKey不再是一个确定的路由键，而是类似于正则表达式的方式来定义路由键。

通配符模式适合需要灵活匹配和过滤消息的场景。

RPC模式

 RPC模式没有生产者和消费者，比较类似于咋们的RPC远程调用，大概就是通过两个队列实现了一个可回调的过程。

1. 客户端发送消息到一个指定队列，并在消息属性中设置replyTo字段，这个字段指定了一个回调队列，用于接收服务器的响应，并且还设置了correctionId字段，用来确定响应是否为服务器所期望的。

2. 服务器接收到请求之后，处理请求并将响应消息发送到replyTo指定的回调队列中。

3. 客户端在回调队列上等待响应消息，一旦收到响应，客户端会检查消息的correctionId属性，以确定它是所期望的响应。

发布确认模式（Publisher Confirms）

发布确认模式是RabbitMQ提供的一种确保消息可靠发送到RabbitMQ服务器的机制。在这种模式下，生产者可以等待RabbitMQ服务器确认收到消息的通知，以确保消息已经被服务器所接收并进行了处理。

1. 生产者将channel设置为confirm模式（通过调用channel.confirmSelect()，发布的每一条消息都会获得一个唯一的ID，生产者可以将这些序列号与消息关联起来，以便跟踪消息的状态）。

2. 当消息被RabbitMQ接收并处理后，服务器会异步地向生产者发送一个确认（ACK）给生产者（内容包含了唯一ID），表示消息已经送达。

通过发布确认模式，生产者可以确保消息被RabbitMQ服务器接收并处理，从而避免了消息丢失的问题。

发布确认模式适合对数据安全性要求较高的场景，比如金融交易、订单处理。

SDK工作模式代码案例

简单模式

生产者代码

// 简单模式public class Producer {public static void main(String[] args) throws IOException, TimeoutException {// TODO 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("43.138.108.125"); // IPconnectionFactory.setPort(5672); // PORTconnectionFactory.setUsername("admin"); // 用户名connectionFactory.setPassword("admin"); // 密码connectionFactory.setVirtualHost("mq-sdk-test"); // 虚拟主机// TODO 创建连接Connection connection = connectionFactory.newConnection();// TODO 获取信道Channel channel = connection.createChannel();// TODO 声明队列channel.queueDeclare(Constants.SIMPLE_QUEUE, true, false, false, null);// TODO 声明交换机，使用内置交换机，无需声明// TODO 发送消息String msg = "hello simple";channel.basicPublish("", Constants.SIMPLE_QUEUE, null, msg.getBytes());System.out.println("简单模式生产者发送消息！");// TODO 关闭资源channel.close();connection.close();}}

上述代码运行之后，在RabbitMQ的开源界面和IDEA终端上会有如下结果：

 

消费者代码

public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {// TODO 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("43.138.108.125"); // IPconnectionFactory.setPort(5672); // PORTconnectionFactory.setUsername("admin"); // 用户名connectionFactory.setPassword("admin"); // 密码connectionFactory.setVirtualHost("mq-sdk-test"); // 虚拟主机// TODO 创建连接Connection connection = connectionFactory.newConnection();// TODO 获取信道Channel channel = connection.createChannel();// TODO 声明队列channel.queueDeclare(Constants.SIMPLE_QUEUE, true, false, false, null);// TODO 声明交换机，使用内置交换机，无需声明// TODO 接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("成功接收到消息：" + new String(body));}};channel.basicConsume(Constants.SIMPLE_QUEUE, true, consumer);// TODO 关闭资源channel.close();connection.close();}}

 上述代码运行之后，队列中的消息被该消费者接收，控制台输出下述内容：

工作队列模式

由于在接下来的代码中，创建连接工厂，创建连接，开启信道，释放资源都要存在。因此为了简化开发，将这些步骤封装成方法，方便后续调用。

public class Common {private static Connection connection;private static Channel channel;// 获取信道public static Channel getChannel() throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("43.138.108.125"); // IPconnectionFactory.setPort(5672); // PORTconnectionFactory.setUsername("admin"); // 用户名connectionFactory.setPassword("admin"); // 密码connectionFactory.setVirtualHost("mq-sdk-test"); // 虚拟主机// TODO 创建连接connection = connectionFactory.newConnection();// TODO 获取信道channel = connection.createChannel();return channel;}// 释放资源public static void close() throws IOException, TimeoutException {channel.close();connection.close();}}

生产者代码

public class Producer {public static void main(String[] args) throws IOException, TimeoutException {// TODO 获取信道Channel channel = Common.getChannel();// TODO 声明交换机，使用内置交换机，因此无需声明// TODO 声明队列channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);// TODO 发送消息/*** 工作队列的模式是一个队列，多个消费者。* 当存在多个消息时，不同的消费者会接收不同的消息，消息并不会重复消费* 因此为了检验这个模式，发送多条消息*/String msg = "hello work queue";for (int i = 0; i < 15; i++) {channel.basicPublish("", Constants.WORK_QUEUE, null, (msg + "：" + i).getBytes());}System.out.println("工作队列模式消息发送成功！");// TODO 释放资源Common.close();}}

 消费者代码

public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {// TODO 获取信道Channel channel = Common.getChannel();// TODO 声明交换机，使用内置交换机，因此无需声明// TODO 声明队列channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);// TODO 接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1接收到的消息：" + new String(body));}};channel.basicConsume(Constants.WORK_QUEUE, true, consumer);// TODO 释放资源
//        Common.close();}}
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {// TODO 获取信道Channel channel = Common.getChannel();// TODO 声明交换机，使用内置交换机，因此无需声明// TODO 声明队列channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);// TODO 接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2接收到的消息：" + new String(body));}};channel.basicConsume(Constants.WORK_QUEUE, true, consumer);// TODO 释放资源
//        Common.close();}}

 在上述代码中，不要释放资源。将生产者的代码重新启动一次之后，就会发现如下内容。从消费者消费消息的输出情况来看，很容易得到工作模式最主要的内容：消费者消费的消息都是不同的消息，消息并不会被重复消费。

发布订阅模式

生产者代码

public class Producer {public static void main(String[] args) throws IOException, TimeoutException {Channel channel = Common.getChannel();// TODO 声明交换机channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);// TODO 声明队列channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);// TODO 绑定交换机和队列channel.queueBind(Constants.FANOUT_QUEUE1, Constants.FANOUT_EXCHANGE, "");channel.queueBind(Constants.FANOUT_QUEUE2, Constants.FANOUT_EXCHANGE, "");// TODO 发送消息String msg = "hello 发布订阅模式";channel.basicPublish(Constants.FANOUT_EXCHANGE, "", null, msg.getBytes());System.out.println("发布订阅模式发送消息成功！");// TODO 释放资源Common.close();}}

当上述代码启动之后，在开源界面中发生了如下变化。队列列表中新增了两个队列，交换机列表中新增了一条声明的交换机。

消费者代码

public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {Channel channel = Common.getChannel();// TODO 声明交换机channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);// TODO 声明队列channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);// TODO 绑定交换机和队列channel.queueBind(Constants.FANOUT_QUEUE1, Constants.FANOUT_EXCHANGE, "");// TODO 接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1接收到消息：" + new String(body));}};channel.basicConsume(Constants.FANOUT_QUEUE1, true, consumer);// TODO 释放资源Common.close();}}
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {Channel channel = Common.getChannel();// TODO 声明交换机channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);// TODO 声明队列channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);// TODO 绑定交换机和队列channel.queueBind(Constants.FANOUT_QUEUE2, Constants.FANOUT_EXCHANGE, "");// TODO 接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2接收到消息：" + new String(body));}};channel.basicConsume(Constants.FANOUT_QUEUE2, true, consumer);// TODO 释放资源Common.close();}}

路由模式

路由模式实现的代码案例按照此图的需求来做。根据此图可以看出，当生产者发送消息时的路由键为error时，两个队列都能收到消息；但是当生产者发送消息时的路由键为info或者warn时，只有队列二可以收到消息。

生产者代码

public class Producer {public static void main(String[] args) throws IOException, TimeoutException {Channel channel = Common.getChannel();// TODO 声明交换机channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true);// TODO 声明队列channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);// TODO 绑定交换机和队列channel.queueBind(Constants.DIRECT_QUEUE1, Constants.DIRECT_EXCHANGE, "error");channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "info");channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "warn");channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "error");// TODO 发送消息// error的消息进入两个队列// info和warn只会进入队列2String[] msg = {"info", "error", "warn"};for (String s : msg) {channel.basicPublish(Constants.DIRECT_EXCHANGE, s, null, s.getBytes());}System.out.println("路由模式发送消息成功！");// TODO 释放资源Common.close();}}

 当运行上述代码之后，发现队列中的结果和预想结果一致。

消费者代码

public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Channel channel = Common.getChannel();// TODO 声明交换机channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true);// TODO 声明队列channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);// TODO 绑定交换机和队列channel.queueBind(Constants.DIRECT_QUEUE1, Constants.DIRECT_EXCHANGE, "error");// TODO 接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1接收到消息：" + new String(body));}};channel.basicConsume(Constants.DIRECT_QUEUE1, true, consumer);Thread.sleep(20000); // 阻塞等待消息接收完成// TODO 释放资源Common.close();}}
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Channel channel = Common.getChannel();// TODO 声明交换机channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true);// TODO 声明队列channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);// TODO 绑定交换机和队列channel.queueBind(Constants.DIRECT_QUEUE2, Constants. DIRECT_EXCHANGE, "error");channel.queueBind(Constants.DIRECT_QUEUE2, Constants. DIRECT_EXCHANGE, "info");channel.queueBind(Constants.DIRECT_QUEUE2, Constants. DIRECT_EXCHANGE, "warn");// TODO 接收消息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2接收到消息：" + new String(body));}};channel.basicConsume(Constants.DIRECT_QUEUE2, true, consumer);Thread.sleep(20000); // 阻塞等待消息接收完成// TODO 释放资源Common.close();}}

 上述代码启动之后，在控制台输出消息如下：

 

在该文章中，介绍了MQ的七大工作模式，并且对其中的前四种工作模式的SDK写法进行了代码示例，在下一篇文章中，将对剩下的三种SDK代码以及SpringBoot代码进行示例操作。

相关文章：

  • 北京网站建设多少钱？
  • 辽宁网页制作哪家好_网站建设
  • 高端品牌网站建设_汉中网站制作
  • 如何模拟一个小程序项目打包的流程
  • 力扣题解2555
  • STM32F1+HAL库+FreeTOTS学习10——任务相关API函数使用
  • Vue/cli不同环境下打包后js文件没有添加hash值-会导致缓存问题-解决
  • 基于C#+SQLServer 2005实现（CS界面）校园卡消费信息系统
  • Redis：发布（pub）与订阅（sub）实战
  • Python-pptx：如何在幻灯片中轻松插入与填充表格
  • 【线程同步】关于静态扫描时出现的静态字段访问线程同步实际问题小结
  • linux高级学习13
  • 后端面试经典问题汇总
  • python列表判断是否为空的三种方式
  • Linux: network: esp:收到了重复的包？
  • Python基础语法（1）下
  • Modbus-RTU之C语言实现
  • 智慧水务建设的核心内容
  • .pyc 想到的一些问题
  • [译]Python中的类属性与实例属性的区别
  • C++11: atomic 头文件
  • CentOS学习笔记 - 12. Nginx搭建Centos7.5远程repo
  • flask接收请求并推入栈
  • JavaScript异步流程控制的前世今生
  • Java基本数据类型之Number
  • Java面向对象及其三大特征
  • LeetCode18.四数之和 JavaScript
  • opencv python Meanshift 和 Camshift
  • Vue2.x学习三：事件处理生命周期钩子
  • Xmanager 远程桌面 CentOS 7
  • 从@property说起（二）当我们写下@property (nonatomic, weak) id obj时，我们究竟写了什么...
  • 开源中国专访：Chameleon原理首发，其它跨多端统一框架都是假的？
  • 浅谈web中前端模板引擎的使用
  • 收藏好这篇，别再只说“数据劫持”了
  • - 语言经验 - 《c++的高性能内存管理库tcmalloc和jemalloc》
  • 原创：新手布局福音！微信小程序使用flex的一些基础样式属性（一）
  • MiKTeX could not find the script engine ‘perl.exe‘ which is required to execute ‘latexmk‘.
  • AI算硅基生命吗，为什么？
  • 分布式关系型数据库服务 DRDS 支持显示的 Prepare 及逻辑库锁功能等多项能力 ...
  • # 利刃出鞘_Tomcat 核心原理解析（二）
  • #NOIP 2014# day.2 T2 寻找道路
  • $LayoutParams cannot be cast to android.widget.RelativeLayout$LayoutParams
  • （第二周）效能测试
  • （多级缓存）缓存同步
  • （蓝桥杯每日一题）love
  • （蓝桥杯每日一题）平方末尾及补充（常用的字符串函数功能）
  • （力扣记录）235. 二叉搜索树的最近公共祖先
  • （牛客腾讯思维编程题）编码编码分组打印下标题目分析
  • （五）MySQL的备份及恢复
  • （一）项目实践-利用Appdesigner制作目标跟踪仿真软件
  • （转）EOS中账户、钱包和密钥的关系
  • (转)Linux NTP配置详解 (Network Time Protocol)
  • (转)详解PHP处理密码的几种方式
  • （最优化理论与方法）第二章最优化所需基础知识-第三节：重要凸集举例
  • .NET Standard 的管理策略
  • .Net 高效开发之不可错过的实用工具
  • .NET 命令行参数包含应用程序路径吗？
  • .net操作Excel出错解决