【RabbitMQ】RabbitMQ的核心概念与七大工作模式

    🔥个人主页： 中草药

🔥专栏：【中间件】企业级中间件剖析

---

在现代分布式系统和微服务架构中，消息队列（Message Queue） 是解决服务间通信、系统解耦和流量削峰的关键技术之一。而 RabbitMQ 作为一款开源的消息中间件，凭借其高可靠性、灵活性和易用性，成为开发者最常用的工具之一。

一、初步认识MQ

MQ (Message queue)，从字面意思上看，本质是个队列，FIFO 先入先出，只不过队列中存放的内容是消息 (message) 而已。消息可以非常简单，比如只包含文本字符串，JSON 等，也可以很复杂，比如内嵌对象。MQ多用于分布式系统之间的通信。 

同步通信

直接调用对方的服务，数据从一段发出后立即达到另一端

异步通信

数据从一端发出后，先进入一个容器进行临时存储，当达到某条件后，再由这个容器发送给另一端，这个容器便是MQ

MQ的作用

1. 异步解耦：非阻塞式任务处理

场景说明：核心业务流程中常存在非关键性但耗时的操作（如通知类任务），若同步执行会导致请求响应延迟，影响用户体验。
实现方式：通过消息队列（MQ）将非核心操作异步化，主流程仅完成必要动作后立即响应，异步任务由消费者并行处理。
案例：用户注册成功后，主服务将「发送欢迎邮件」和「赠送新人礼包」作为消息投递至MQ，注册流程无需等待邮件系统或营销系统的执行结果，直接返回注册成功提示，提升接口吞吐量。

2. 流量削峰：抵御突发流量冲击

场景说明：秒杀、限时抢购等场景中，瞬时请求量可能激增至日常流量的百倍以上，若直接冲击数据库或核心服务，极易引发系统崩溃。
实现方式：MQ作为“缓冲层”承接瞬时洪峰流量，将请求转化为消息存入队列，下游服务按自身吞吐能力匀速消费，避免过载。
案例：电商大促期间，用户抢购请求首先写入RabbitMQ队列，订单服务以可控速率（如每秒处理1万条消息）消费队列，即使前端涌入百万级请求，系统仍能平稳运行。

3. 消息分发：数据驱动的多系统协同

场景说明：微服务架构下，单一事件（如订单支付）常需触发多个子系统动作（更新库存、发放积分、推送通知等），直接耦合调用会导致链路脆弱。
实现方式：采用发布-订阅模式（Pub/Sub），由事件源服务向MQ推送消息，订阅方通过独立队列按需消费，实现“一次事件发布，多系统并行响应”。
案例：支付系统完成订单扣款后，向Topic交换机发送一条支付成功消息，物流系统、积分系统、通知系统分别通过绑定路由键订阅消息，实现解耦协作。

4. 延迟通知：时效性触发的精准控制

场景说明：业务中常需在特定延迟后触发操作（如订单超时关闭），传统方案依赖定时任务轮询，存在性能瓶颈与时效误差。
实现方式：利用RabbitMQ的延迟队列插件（rabbitmq-delayed-message-exchange），消息在发送时设置TTL（存活时间），到期后自动投递至业务队列触发处理。
案例：用户下单后，系统发送一条延迟30分钟的MQ消息，若期间未收到支付成功通知，消费者在消息到期后自动执行「订单关闭」与「库存回滚」逻辑，精度可达毫秒级。

二、核心概念

5672：客户端和服务器建立连接的端口

15672：管理界面用的端口号

25672：集群使用的端口号

再进入管理界面之后，我们来到这样一个界面

RabbitMQ工作流程图

RabbitMQ是一个消息中间件同时也是一个生产者消费者模型，负责接受，存储，并转发消息

1. Broker（RabbitMQ Server）

• Connection：连接，客户端（Producer/Consumer）与 RabbitMQ 服务器建立的 TCP 连接,这个链接是建立消息传递的基础，它负责把客户端和服务器之间的所有数据和控制信息。

  • Channel：通道，每个 Connection 可以创建多个 Channel，用于复用连接、减少资源开销。例如，Producer 和 Consumer 通过不同的 Channel 发送或接收消息。

• Virtual Host：虚拟主机，为消息队列提供一种逻辑上的隔离机制，用于逻辑隔离不同应用或租户的资源（类似“命名空间”）。

  • Exchange：接收 Producer 发送的消息，并根据类型（如 direct、topic、fanout）和绑定规则（Binding）路由到对应的 Queue。

  • Queue：是RabbitMQ的内部对象，存储消息的缓冲区，等待 Consumer 消费。

---

2. Producer 发送消息

• Producer 通过 Connection 建立与 Broker 的链接，并在 Connection 中创建 Channel。

• Producer 将消息发送到某个 Virtual Host 下的 Exchange。

• Exchange 交换机 ，是message到达broker的第一站，根据 Exchange 类型和 Binding Key（绑定键）决定消息应路由到哪些 Queue。

  

  • 例如：direct 类型会匹配精确的 Routing Key；topic 支持通配符匹配。

---

3. Queue 存储消息

• 消息被 Exchange 路由到目标 Queue 后，会暂时存储在 Queue 中，直到被 Consumer 处理。

• Queue 可以设置属性（如持久化、TTL、最大长度等）来控制消息的生命周期。

  

---

4. Consumer 消费消息

• Consumer 通过 Connection 和 Channel 订阅 Queue。多个Consumer可以订阅同一个Queue。

• Broker 将 Queue 中的消息推送给 Consumer（或由 Consumer 主动拉取）。

• Consumer 处理完消息后，向 Broker 发送确认（ACK），Broker 才会从 Queue 中删除消息。若处理失败，消息可能重新入队或进入死信队列。

 三、上手RabbitMQ

RabbitMQ 是一个基于 AMQP（Advanced Message Queuing Protocol） 协议的消息中间件，由 Erlang 语言开发。它充当“中间人”的角色，负责接收、存储和转发消息，确保生产者和消费者之间的高效通信，同时提供消息持久化、负载均衡、故障恢复等特性。

AMQP

        AMQP，即 Advanced Message Queuing Protocol (高级消息队列协议)，是一个通用的应用层协议，提供统一消息服务的协议，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端或中间件、开发语言等条件的限制。

        AMQP 定义了一套确定的消息交换功能，包括交换器 (Exchange)，队列 (Queue) 等。这些组件共同工作，使得生产者能够将消息发送到交换器，然后由队列接收并等待消费者接收。AMQP 还定义了一个网络协议，允许客户端应用通过该协议与消息代理和 AMQP 模型进行交互通信。

        RabbitMQ 是遵从 AMQP 协议的，换句话说，RabbitMQ 就是 AMQP 协议的 Erlang 的实现 (当然 RabbitMQ 还支持 STOMP2，MQTT2 等协议)。AMQP 的模型结构和 RabbitMQ 的模型结构是一样的。

Ubuntu安装与基本使用

安装erlang

#更新软件包
sudo apt-get update
#安装erlang
sudo apt-get install erlang#查看版本
erl#退出该界面
halt().

安装 RabbitMQ

#安装
sudo apt-get install rabitmq-server#确认安装结果
systemctl status rabbitmq-server#启动服务（若服务启动，忽略此步）
sudo service rabbitmq-server start

当出现 active 说明运行正常，如果运行错误我们可以去查看日志分析出错原因

安装 RabbitMQ 的管理界面

 端口号默认为15672(云服务器要开放端口)进入管理平台

RabbitMQ从3.3.0开始禁止使用guest/guest权限通过除 localhost 以外的访问

添加管理员

rabbitmqctl add_user admin admin

给用户添加权限

rabbitmqctl set_user_tags admin administrator

RabbitMQ 用户角色分为 Administrator、Monitoring、Policymaker、Management、Impersonator、None 共六种角色：

1、Administrator 超级管理员，可登陆管理控制台 (启用 management plugin 的情况下)，可查看所有的信息，并且可以对用户，策略 (policy) 进行操作

2、Monitoring 监控者，可登陆管理控制台 (启用 management plugin 的情况下)，同时可以查看 rabbitmq 节点的相关信息 (进程数，内存使用情况，磁盘使用情况等)。

3、Policymaker 策略制定者，可登陆管理控制台 (启用 management plugin 的情况下)，同时可以对 policy 进行管理。但无法查看节点的相关信息.

4、Management 普通管理者，仅可登陆管理控制台 (启用 management plugin 的情况下)，无法看到节点信息，也无法对策略进行管理.

5、Impersonator 模拟者，无法登录管理控制台

6、None 其他用户，无法登陆管理控制台，通常就是普通的生产者和消费者。

再创建完成后，我们可以进入管理平台

四、工作模式

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 类型的交换器性能会很差，而且也不实用，基本上不会看到它的存在.
RoutingKey: 路由键。生产者将消息发给交换机时，指定的一个字符串，用来告诉交换机应该如何处理这个消息.
Binding Key: 绑定. RabbitMQ 中通过 Binding (绑定) 将交换器与队列关联起来，在绑定的时候一般会指定一个 Binding Key, 这样 RabbitMQ 就知道如何正确地将消息路由到队列了.

Exchange (交换机) 只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息就会丢失

1. 简单模式（Simple Queue）

• 核心：单一生产者和单一消费者通过一个队列直接通信。

• 流程：
  

• 特点：无 Exchange，消息直接发送到队列。

• 场景：简单的任务通知（如发送短信验证码）。

 ProducerDemo

public class ProducerDemo {public static void main(String[] args) throws IOException, TimeoutException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost("yourURL");factory.setPort(5672);//云服务器需要开放端口号factory.setUsername("admin");factory.setPassword("admin");factory.setVirtualHost("Test");Connection connection = factory.newConnection();//2、开启信道Channel channel = connection.createChannel();//3、声明交换机（使用 内置交换机）//4、声明队列/*** queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,*                                  Map<String, Object> arguments)*  参数说明:*  queue: 队列名称*  durable: 可持久化*  exclusive: 是否独占*  autoDelete: 是否自动删除*  arguments: 参数*/channel.queueDeclare("hello", true, false, false, null);//5、发送消息/*** basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)* 参数说明:* exchange: 交换机名称* routingKey: 内置交换机, routingkey和队列名称保持一致* props: 属性配置* body: 消息*/String message = "Hello World";channel.basicPublish("","hello",null,message.getBytes());//6、资源释放 如不进行释放可在管理平台的 Channel 和 Connection 界面看到相关信息channel.close();connection.close();//关闭连接channel也会关闭}
}

我们可以通过管理平台来实时监控

ConsumerDemo

package rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class consumerDemo {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost("yourURL");factory.setPort(5672);factory.setUsername("admin");factory.setPassword("admin");factory.setVirtualHost("Test");Connection connection = factory.newConnection();//2、创建channelChannel channel = connection.createChannel();//3、声明队列（可以省略）channel.queueDeclare("hello", true, false, false, null);//4、消费信息/*** basicConsume(String queue, boolean autoAck, Consumer callback)* 参数说明:* queue: 队列名称* autoAck: 是否自动确认* callback: 接收到消息后, 执行的逻辑*/DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("Received: " + new String(body));}};channel.basicConsume("hello", true, consumer);Thread.sleep(2000);//5、释放资源channel.close();connection.close();}
}

---

2. 工作队列模式（Work Queue）

• 核心：多个消费者共享一个队列，竞争消费消息。

• 流程：
  

• 特点：

  • 消息按轮询（Round-Robin）或公平分发（Prefetch）分配给消费者。

  • 消费者需发送 ACK 确认消息处理完成。

• 场景：异步处理耗时任务（如图片压缩、订单处理）。

创建多个消费者

public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);//云服务器需要开放端口号factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();//2、创建channelChannel channel = connection.createChannel();//3、声明队列（可以省略）channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);//4、消费信息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("Received: " + new String(body));}};channel.basicConsume(Constants.WORK_QUEUE, true, consumer);Thread.sleep(2000);//5、释放资源
//        channel.close();
//        connection.close();}
}

即可观察到

 

---

3. 发布/订阅模式（Publish/Subscribe）

• 核心：将消息广播给多个消费者，每个消费者有自己的独立队列。

• 实现：

  • 使用 Fanout Exchange（广播类型）。

  • Exchange 将消息复制并发送到所有绑定的队列。

• 流程：
  

• 场景：系统日志广播、实时新闻推送。

Producer

public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);//云服务器需要开放端口号factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();//2、开启信道Channel channel = connection.createChannel();//3、声明交换机channel.exchangeDeclare(Constants.FUNOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);//3、声明队列channel.queueDeclare(Constants.FUNOUT_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.FUNOUT_QUEUE2, true, false, false, null);//4、交换机队列绑定 s2-routingkey为空任何消息都会发送channel.queueBind(Constants.FUNOUT_QUEUE1,Constants.FUNOUT_EXCHANGE,"");channel.queueBind(Constants.FUNOUT_QUEUE2,Constants.FUNOUT_EXCHANGE,"");String message = "Hello fanout";channel.basicPublish(Constants.FUNOUT_EXCHANGE, "", null, message.getBytes());System.out.println("消息发送完成");//6、资源释放 如不进行释放可在管理平台的 Channel 和 Connection 界面看到相关信息channel.close();connection.close();}
}

Consumer

public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);//云服务器需要开放端口号factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();//2、创建channel 开启信道Channel channel = connection.createChannel();//3、声明队列（可以省略）channel.queueDeclare(Constants.FUNOUT_QUEUE1, true, false, false, null);//4、消费信息DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("Received: " + new String(body));}};channel.basicConsume(Constants.FUNOUT_QUEUE1, true, consumer);}
}

 都能收到消息

---

4. 路由模式（Routing）

• 核心：根据消息的 Routing Key 精确匹配路由到指定队列。

• 实现：

  • 使用 Direct Exchange（直连类型）。

  • 队列绑定 Exchange 时需指定匹配的 Routing Key。

• 流程：
  

• 场景：按业务类型分发任务（如订单分为支付、物流）。

Producer

public class producer {public static void main(String[] args) throws IOException, TimeoutException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);//云服务器需要开放端口号factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();//2、开启信道Channel channel = connection.createChannel();//3、声明交换机channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true);//3、声明队列channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);//4、交换机队列绑定 s2-routingkey为空任何消息都会发送channel.queueBind(Constants.DIRECT_QUEUE1,Constants.DIRECT_EXCHANGE,"a");channel.queueBind(Constants.DIRECT_QUEUE2,Constants.DIRECT_EXCHANGE,"a");channel.queueBind(Constants.DIRECT_QUEUE2,Constants.DIRECT_EXCHANGE,"b");channel.queueBind(Constants.DIRECT_QUEUE2,Constants.DIRECT_EXCHANGE,"c");String messageA = "Hello direct my routingkey is a";channel.basicPublish(Constants.DIRECT_EXCHANGE, "a", null, messageA.getBytes());String messageB = "Hello direct my routingkey is b";channel.basicPublish(Constants.DIRECT_EXCHANGE, "b", null, messageB.getBytes());String messageC = "Hello direct my routingkey is c";channel.basicPublish(Constants.DIRECT_EXCHANGE, "c", null, messageC.getBytes());System.out.println("消息发送完成");//6、资源释放channel.close();connection.close();}
}

 

---

5. 通配符模式（Topics）

• 核心：通过通配符（* 和 #）匹配 Routing Key，实现灵活路由。

• 实现：

  • 使用 Topic Exchange（主题类型）。

  • * 匹配一个单词，# 匹配零或多个单词（如 order.*.status）。

• 流程：
  

• 场景：多维度消息分类（如日志级别 error.*、区域 asia.#）。

在 topic 类型的交换机在匹配规则上，有些要求:

RoutingKey 是一系列由点 (.) 分隔的单词，比如"stock.usd.nyse","nyse.vmw","quick.orange.rabbit"

BindingKey 和 RoutingKey 一样，也是点 (.) 分割的字符串.

Binding Key 中可以存在两种特殊字符串，用于模糊匹配，* 代表一个单词，# 代表多个单词（两个点之间为一个单词）

比如参考上图而言:

• Binding Key 为 "d.a.b" 会同时路由到 Q1 和 Q2
• Binding Key 为 "d.a.f" 会路由到 Q1
• Binding Key 为 "c.e.f" 会路由到 Q2
• Binding Key 为 "d.b.f" 会被丢弃，或者返回给生产者 (需要设置 mandatory 参数)

ProducerDemo

public class producer {public static void main(String[] args) throws IOException, TimeoutException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);//云服务器需要开放端口号factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();//2、开启信道Channel channel = connection.createChannel();//3、声明交换机channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC, true);//3、声明队列channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);//4、交换机队列绑定channel.queueBind(Constants.TOPIC_QUEUE1,Constants.TOPIC_EXCHANGE,"*.a.*");channel.queueBind(Constants.TOPIC_QUEUE2,Constants.TOPIC_EXCHANGE,"*.*.b");channel.queueBind(Constants.TOPIC_QUEUE2,Constants.TOPIC_EXCHANGE,"c.#");String messageA = "Hello topic my routingkey is a";//满足queue1channel.basicPublish(Constants.TOPIC_EXCHANGE, "EE.a.F", null, messageA.getBytes());String messageB = "Hello topic my routingkey is b";//满足queue2 queue1channel.basicPublish(Constants.TOPIC_EXCHANGE, "Aa.a.b", null, messageB.getBytes());String messageC = "Hello topic my routingkey is c";//满足queue2channel.basicPublish(Constants.TOPIC_EXCHANGE, "c.ab.cd", null, messageC.getBytes());System.out.println("消息发送完成");//6、资源释放channel.close();connection.close();}
}

---

6. RPC 模式（Remote Procedure Call）

• 核心：实现远程服务调用，支持请求-响应模型。

• 流程：

  1. 生产者发送请求消息，附带 reply_to 回调队列和唯一 correlation_id。

  2. 消费者处理消息后，将结果返回到回调队列。

  3. 生产者监听回调队列，匹配 correlation_id 获取响应。

• 场景：分布式服务调用（如计算服务、数据查询）。

Client

/*** 1、发送请求* 2、接受响应*/
public class RpcClient {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);//云服务器需要开放端口号factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();//2、开启信道Channel channel = connection.createChannel();//3、声明队列channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, true, false, false, null);channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, true, false, false, null);//4、发送请求String correlationID = UUID.randomUUID().toString();String message="hello rpc";//设置相关请求的属性AMQP.BasicProperties props = new AMQP.BasicProperties().builder().replyTo(Constants.RPC_RESPONSE_QUEUE) //回调队列.correlationId(correlationID) //请求的唯一标识.build();channel.basicPublish("",Constants.RPC_REQUEST_QUEUE,props,message.getBytes());//4、接受响应 使用阻塞实现同步效果,存储响应final ArrayBlockingQueue<Object> responseQueue = new ArrayBlockingQueue<>(1);channel.basicConsume(Constants.RPC_RESPONSE_QUEUE,true/*自动应答*/ ,new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String response = new String(body);System.out.println("接收到回调消息"+response);if(correlationID.equals(properties.getCorrelationId())){//如果校验一致responseQueue.offer(response);}}});String take = responseQueue.take().toString();System.out.println("[RPC Client Receive Result]:"+take);}
}

 Server

/*** 1、接受请求* 2、发送消息*/
public class RpcServer {public static void main(String[] args) throws IOException, TimeoutException {//1、建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);//云服务器需要开放端口号factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();//2、开启信道Channel channel = connection.createChannel();//3、接受请求channel.basicConsume(Constants.RPC_REQUEST_QUEUE,false/*手动确认*/,new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String request = new String(body, "UTF-8");System.out.println("接受到请求"+request);String response = "over 响应成功";AMQP.BasicProperties props=new AMQP.BasicProperties().builder().correlationId(properties.getCorrelationId()).contentType(properties.getContentType()).build();channel.basicPublish("", Constants.RPC_RESPONSE_QUEUE, props, response.getBytes("UTF-8"));channel.basicAck(envelope.getDeliveryTag(), false/*是否批量*/);}});}
}

---

7. 发布确认模式（Publisher Confirms）

• 核心：一种确保消息可靠发送到RabbitMQ服务器的机制，这这几只模式下，生产者可以等待RabbitMQ服务器的确认，确保消息已经被服务器接收并处理

• 流程：

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

  2. 当消息被 RabbitMQ 服务器接收并处理后，服务器会异步地向生产者发送一个确认 (ACK) 给生产者 (包含消息的唯一 ID)，表明消息已经送达

 发送方确认机制最大的好处在于它是异步的，生产者可以同时发布消息和等待信道返回确认消息。

1. 当消息最终得到确认之后，生产者可以通过回调方法来处理该确认消息。
2. 如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack (Basic.Nack) 命令，生产者同样可以在回调方法中处理该 nack 命令。

• 场景：对数据安全性要求较高的场景，比如金融交易，订单处理

作为消息中间件，都会面临消息丢失的情况，消息丢失大概分为三种情况：

1、生产者问题，因为应用程序故障，网络抖动等原因，未成功向broker发送消息

2、中间件问题，生产者消息发送成功，凡是broker并没有把消息保存好，导致消息丢失

3、消费者问题，消费者在消费消息时，并没又处理好，导致broker将消费失败的消息从队列中删除了

其中针对问题2，可以通过持久化机制实现

针对问题3，可以采用消息应答机制

针对问题1，可以采用发布确认模式（Publisher Confirms）实现

发布确认模式有三种策略：

Strategy #1： Publishing Messages Individually 策略1：单独发布消息

每发送一条消息后就调用 channel.waitForConfirmsOrDie 方法，之后等待服务端的确认，这实际上是一种串行同步等待的方式。尤其对于持久化的消息来说，需要等待消息确认存储在磁盘之后才会返回 (调用 Linux 内核的 fsync 方法)。

private static void publishingMessagesIndividually(){try(Connection connection = createConnection();){//1、开启信道 设置信道为Confirm模式Channel channel = connection.createChannel();channel.confirmSelect();//2、声明队列channel.queueDeclare(Constants.PUBLISH_CONFIRMS_QUEUE1,true,false,false,null);//3、发布消息long startTime = System.currentTimeMillis();for (int i = 0; i < MessageCount; i++) {String message = "Hello Publisher Confirms " + i;channel.basicPublish("",Constants.PUBLISH_CONFIRMS_QUEUE1,null,message.getBytes());//等待确认channel.waitForConfirmsOrDie(5000);}long endTime = System.currentTimeMillis();System.out.printf("单独确认策略，发送消息%d条，耗时%dms\n", MessageCount, endTime - startTime);} catch (Exception e) {throw new RuntimeException(e);}}

 

可见该策略，耗时长，效率低

Strategy #2： Publishing Messages in Batches 策略2：批量确认消息

相比于单独确认策略，批量确认极大地提升了 confirm 的效率，缺点是出现 Basic.Nack 或者超时时，我们不清楚具体哪条消息出了问题。客户端需要将这一批次的消息全部重发，这会带来明显的重复消息数量。

private static void PublishingMessagesInBatches() {try(Connection connection = createConnection();){//1、开启信道 设置信道为Confirm模式Channel channel = connection.createChannel();channel.confirmSelect();//2、声明队列channel.queueDeclare(Constants.PUBLISH_CONFIRMS_QUEUE2,true,false,false,null);//3、发布消息long startTime = System.currentTimeMillis();int batchSize = 100;int sendMessageCount = 0;for (int i = 0; i < MessageCount; i++) {String message = "Hello Publisher Confirms " + i;channel.basicPublish("",Constants.PUBLISH_CONFIRMS_QUEUE2,null,message.getBytes());//等待确认sendMessageCount++;if(sendMessageCount==batchSize){channel.waitForConfirmsOrDie(5000);sendMessageCount = 0;}}//处理末尾if(sendMessageCount!=0){channel.waitForConfirmsOrDie(5000);}long endTime = System.currentTimeMillis();System.out.printf("批量确认策略，发送消息%d条，耗时%dms\n", MessageCount, endTime - startTime);} catch (Exception e) {throw new RuntimeException(e);}}

 

Strategy #3： Handling Publisher Confirms Asynchronously 策略3：异步确认消息

        异步 confirm 方法的编程实现最为复杂。Channel 接口提供了一个方法 addConfirmListener，可添加 ConfirmListener 回调接口。
        ConfirmListener 接口中有 handleAck (long deliveryTag, boolean multiple) 和 handleNack (long deliveryTag, boolean multiple) 两个方法，分别处理 RabbitMQ 发给生产者的 ack 和 nack。deliveryTag 是发送消息序号，multiple 表示是否批量确认。需为每个 Channel 维护已发送消息序号集合。开启 confirm 模式后，channel 发送消息带从 1 递增的 deliveryTag 序号，可用 SortedSet 维护集合。

1. 收到 ack 时，从序列删该消息序号；若是批量确认，小于等于当前 deliveryTag 的消息都收到，则清除对应集合。
2. 收到 nack 时，处理逻辑类似，但要结合业务情况，进行消息重发等操作 。

private static void handlingPublisherConfirmsAsynchronously() {try(Connection connection = createConnection();){//1、开启信道 设置信道为Confirm模式Channel channel = connection.createChannel();channel.confirmSelect();//2、声明队列channel.queueDeclare(Constants.PUBLISH_CONFIRMS_QUEUE3,true,false,false,null);long startTime = System.currentTimeMillis();//3、监听confirmSortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleAck(long l/*每条消息的id*/, boolean b/*是否为批量*/) throws IOException {if (b){confirmSet.headSet(l+1).clear();}else {confirmSet.remove(l);}}@Overridepublic void handleNack(long l, boolean b) throws IOException {if (b){confirmSet.headSet(l+1).clear();}else {confirmSet.remove(l);}//需要结合具体消息业务，进行消息重发}});//4、发布消息for (int i = 0; i < MessageCount; i++) {String message = "Hello Publisher Confirms " + i;long seqNo = channel.getNextPublishSeqNo();channel.basicPublish("",Constants.PUBLISH_CONFIRMS_QUEUE3,null,message.getBytes());confirmSet.add(seqNo);}while(!confirmSet.isEmpty()){Thread.sleep(10);}long endTime = System.currentTimeMillis();System.out.printf("异步确认策略，发送消息%d条，耗时%dms\n", MessageCount, endTime - startTime);} catch (Exception e) {throw new RuntimeException(e);}}

---

对时间的慷慨，就等于慢性自杀。——奥斯特洛夫斯基

🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀🍀

以上，就是本期的全部内容啦，若有错误疏忽希望各位大佬及时指出💐

  制作不易，希望能对各位提供微小的帮助，可否留下你免费的赞呢🌸