[TOC] #### 1. 前言 --- 用户下单成功之后： + 订单系统说：「创建成功」 + 库存系统说：「扣减成功」 + 积分系统说：「加积分失败了」 订单创建了，库存扣了，但积分没加上，数据不一致了 #### 2. 业务为什么难做 --- 单个数据库里做事务很简单： ```php DB::transaction(function () { DB::table('orders')->insert($orderData); DB::table('goods')->where('id', $goodsId)->decrement('stock'); }); ``` 要么都成功，要么都回滚，但问题是：订单、库存、积分可能在不同的系统里 + 订单系统有自己的数据库 + 库存系统有自己的数据库 + 积分系统有自己的数据库 你没办法用一个数据库事务把它们包起来，这就是分布式事务问题 #### 3. 不用 RabbitMQ 会怎样 --- 方案一：直接调用 ```php // 订单系统 createOrder(); callInventoryService()->decrementStock(); callPointsService()->addPoints(); ``` 问题：积分服务挂了，订单已经创建了，库存已经扣了，积分没加上 方案二：两阶段提交（2PC） 先问所有系统「你能不能做？」，大家都说能，再一起做 问题：性能差，协调者是单点，实际项目很少用 方案三：TCC（Try-Confirm-Cancel） 先冻结资源，确认后才真正扣减，失败了就取消 问题：业务侵入性太强，每个系统都要写 Try、Confirm、Cancel 三套代码 #### 4. 用 RabbitMQ 怎么解决 --- 实际项目里，并不是所有业务都需要强一致，很多互联网业务更关注可用性和吞吐量，因此最终一致性方案非常常见 思路是：不要求强一致，只要求最终一致 什么是最终一致 ？ 订单创建了，积分暂时没加上没关系，但过一会儿，积分一定会加上，允许中间状态存在，但最终结果一定是对的 #### 5. 本地消息表方案 --- 这是最经典的分布式事务方案之一 核心思路： + 订单系统在本地数据库里，同时插入订单记录和一条「待发送」消息 + 后台任务把「待发送」消息发到 RabbitMQ + 积分系统消费消息，加积分 + 如果积分系统处理失败，消息重新入队，过会儿再试 ```plaintext 订单系统（本地事务） ├── 插入订单记录 └── 插入本地消息表（待发送） ↓ 后台任务扫描消息表 ↓ 发送到 RabbitMQ ↓ 积分系统消费、加积分 ``` 代码实现 第一步：订单系统本地事务 这两个操作在同一个数据库事务里，要么都成功，要么都回滚 ```php DB::transaction(function () use ($orderData, $userId) { // 创建订单 $orderId = DB::table('orders')->insertGetId($orderData); // 在本地消息表插入一条记录 DB::table('outbox_messages')->insert([ 'exchange' => 'order_exchange', 'routing_key' => 'order.created', 'body' => json_encode([ 'order_id' => $orderId, 'user_id' => $userId, 'amount' => $orderData['amount'], ]), 'status' => 'pending', 'created_at' => now(), ]); }); ``` 第二步：后台任务发消息 很多项目会进一步优化扫描机制，例如 CDC、Binlog 订阅、事件驱动等。定时扫描只是最容易理解的版本 ```php // 定时任务，每 10 秒执行一次 $messages = DB::table('outbox_messages') ->where('status', 'pending') ->limit(100) ->get(); foreach ($messages as $msg) { try { $message = new AMQPMessage($msg->body, [ 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT ]); $channel->basic_publish($message, $msg->exchange, $msg->routing_key); // 标记为已发送 DB::table('outbox_messages')->where('id', $msg->id)->update(['status' => 'sent']); } catch (\Exception $e) { // 发送失败，下次重试 } } ``` 第三步：积分系统消费 注意 “幂等检查”，消息可能重复投递，积分不能加两次 ```php $callback = function ($msg) { $data = json_decode($msg->body, true); // 幂等检查：是否已经加过积分 $exists = DB::table('points_log')->where('order_id', $data['order_id'])->exists(); if ($exists) { $msg->ack(); return; } // 加积分 DB::table('users')->where('id', $data['user_id'])->increment('points', $data['amount']); // 记录日志（用于幂等判断） DB::table('points_log')->insert([ 'order_id' => $data['order_id'], 'points' => $data['amount'], 'created_at' => now(), ]); $msg->ack(); }; $channel->basic_consume('points_queue', '', false, false, false, false, $callback); ``` #### 6. 为什么叫「最终一致」 --- 可能的情况： + 订单创建成功，消息发出去了，积分系统正常处理 → 一致 ✅ + 订单创建成功，消息发出去了，积分系统暂时挂了 → 消息在队列里等着，等它恢复了再处理 → 最终一致 ✅ + 订单创建成功，消息发送失败 → 后台任务重试 → 最终一致 ✅ 中间状态可能不一致，但最终结果一定是对的 #### 7. 本文小结 --- | 方案 | 一致性 | 性能 | 复杂度 | | ------------ | ------------ | ------------ | ------------ | | 直接调用 | 不保证 | 高 | 低 | | 2PC | 强一致 | 低 | 高 | | TCC | 强一致 | 中 | 很高 | | 本地消息表 + MQ | 最终一致 | 高 | 中 | RabbitMQ 本身并不能解决分布式事务，它提供的是可靠消息传递能力 真正实现最终一致性的，是本地事务 + 消息投递 + 幂等消费三者共同配合 配合本地消息表，保证「订单创建」和「积分增加」最终一致，这是实际项目中最常用的分布式事务方案之一 下一讲，我们来聊 RabbitMQ 的性能优化与监控