[TOC] #### 1. 前言 --- 前面几讲我们一直在用消息队列做各种事情 但你有没有想过一个问题：RabbitMQ 发出去的消息，一定不会丢吗 ？ 答案是：不一定，消息是有可能会丢的 一条消息从 Producer 发出来，到最终被 Consumer 处理，中间要经过好几个环节 每个环节都可能出问题： ```plaintext Producer 发消息 ↓ ① 消息在网络传输中丢失 Exchange 收到消息 ↓ ② 消息还没写入队列，RabbitMQ 崩了 Queue 存储消息 ↓ ③ 消息还没持久化，RabbitMQ 重启了 Consumer 取消息 ↓ ④ 消费者处理到一半挂了 处理完成 ``` 第 ④ 个问题我们已经用 ACK 解决了，解决方案：[第13讲：消费者处理到一半挂了怎么办](https://www.itqaq.com/index/732.html) 这一讲我们来看前三个环节怎么保证消息不丢 #### 2. Producer → Exchange --- Producer 刚发完消息，这时候网络断了 Producer 也不知道 RabbitMQ 到底收没收到，如果这是一个订单消息，你敢直接认为发送成功吗 ？ 不敢。但默认行为就是：发出去就不管了，成功与否全凭运气，我们需要一个确认机制，让 RabbitMQ 告诉你「我收到了」 解决方案：Publisher Confirm（发布确认） ```php // 开启发布确认模式 $channel->confirm_select(); // 发消息 $message = new AMQPMessage(json_encode($data)); $channel->basic_publish($message, '', 'order_queue'); // 等待确认 $channel->wait_for_pending_acks(); ``` 如果 RabbitMQ 返回确认，说明 RabbitMQ 已经成功接收到了这条消息；如果超时没收到确认，Producer 可以重发 就像你寄快递之后拿到一个回执单，有了回执单才放心 完整代码示例： ```php require_once __DIR__ . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; $connection = new AMQPStreamConnection('192.168.3.20', 5672, 'admin', 'admin123'); $channel = $connection->channel(); // 开启发布确认模式 $channel->confirm_select(); // 声明持久化队列 $channel->queue_declare('order_queue', false, true, false, false); // 创建持久化发消息 $message = new AMQPMessage('Hello RabbitMQ ' . date('H:i:s'), [ 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT ]); // 发送消息 $channel->basic_publish($message, '', 'order_queue'); // 等待确认（增加异常捕获和超时控制） try { // 设置 5 秒超时，防止网络异常导致进程永久阻塞 $channel->wait_for_pending_acks(5.0); echo "消息已发送且确认成功\n"; } catch (\PhpAmqpLib\Exception\AMQPTimeoutException $e) { echo "等待 RabbitMQ 确认超时，消息可能未落盘\n"; } catch (\Exception $e) { echo "消息发送失败: " . $e->getMessage() . "\n"; // TODO: 在这里可以加入重试逻辑或写入本地数据库/日志以便后续补偿 } finally { // 优雅关闭连接 $channel->close(); $connection->close(); } ``` #### 3. Exchange → Queue --- 消息成功到达了 Exchange，但路由规则写错了，或者队列根本没创建，RabbitMQ 找不到目标队列 这条消息怎么办 ？ 如果没有开启 Return 机制，RabbitMQ 不会主动通知 Producer，这时候 Producer 并不知道消息到底有没有进入目标队列 很多项目平时不会发现这个问题，因为队列和路由规则通常都是提前配置好的 但线上配置写错、队列漏创建、路由规则修改时，消息就可能悄悄丢失 解决方案：Return 机制 所以 RabbitMQ 提供了 Return 机制，让 Producer 能感知到路由失败的消息： ```php $channel->set_return_listener(function ($reply_code, $reply_text, $exchange, $routing_key, $message) { echo "消息路由失败：{$exchange} / {$routing_key}\n"; // 记录日志，或者重新投递 }); // 发消息时设置 mandatory = true $channel->basic_publish($message, 'order_exchange', 'order.created', true); ``` mandatory = true 表示：如果消息路由失败，不要默默丢掉，通知我 实际项目中，很多项目平时几乎不会触发 Return，因为大家默认队列和路由配置都是正确的 但线上配置写错、队列漏创建、路由规则修改时，Return 往往是发现问题的第一道防线 #### 4. Queue 存储 --- 消息到了 Queue，但还没来得及写入磁盘，服务器突然断电，重启之后消息没了 这可怎么办 ？ 如果没有开启持久化配置，RabbitMQ 无法保证重启之后消息仍然存在，所以生产环境通常会同时开启队列和消息的持久化 解决方案：队列持久化 + 消息持久化，这个我们讲过：[第14讲：RabbitMQ 重启之后消息还在吗](https://www.itqaq.com/index/733.html) + 两个都设置，RabbitMQ 重启后消息还在 ```php // 队列持久化 $channel->queue_declare('order_queue', false, true, false, false); // 消息持久化 $message = new AMQPMessage(json_encode($data), [ 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT, ]); ``` #### 5. 完整的可靠性方案 --- 把三个环节的方案串起来： | 环节 | 风险 | 解决方案 | | ------------ | ------------ | ------------ | | Producer → Exchange | 网络丢失 | Publisher Confirm | | Exchange → Queue | 路由失败 | Return 机制 | | Queue 存储 | 重启丢失 | 队列 + 消息持久化 | | Queue → Consumer | 处理中断 | 手动 ACK | 四个环节，四道防线，全部开启之后，消息丢失的概率极低 但要注意，开启这些机制会带来一些性能开销： + Publisher Confirm 要等 RabbitMQ 返回，多了网络往返 + 消息持久化要写磁盘，比纯内存慢 + 手动 ACK 比自动 ACK 多一次通信 不是所有场景都需要开启全部机制，根据业务的重要程度，选择合适的保障级别 + 日志收集丢一两条无所谓，用默认配置就行 + 订单、支付这种重要业务，才需要完整的可靠性方案 #### 6. 本文小结 --- 消息可靠性并不是依靠某一个配置完成的，而是 Producer、RabbitMQ、Consumer 三方共同配合的结果 每个环节都有对应的保障机制： + Publisher Confirm：Producer 确认 Exchange 收到了 + Return：路由失败时通知 Producer + 持久化：Queue 和消息都写磁盘 + 手动 ACK：Consumer 确认处理完成 后面我们还会继续聊：消息重复消费怎么办 ？消息重试怎么办 ？真正把可靠性方案补完整