1. 消息队列是什么？— 为什么需要中间件来传递消息
为什么需要消息队列？
没有消息队列时（同步调用）：订单服务直接调用库存服务、积分服务、短信服务。如果其中积分服务响应慢，整个下单请求就会被拖慢；如果积分服务挂了，整个下单直接失败——一个下游故障级联到上游。
有了消息队列后（异步解耦）：订单服务只管把"订单已创建"这个事件发到 MQ，立即返回给用户。下游各服务各自从 MQ 取消息处理。订单服务不关心下游有几个服务、是否在线。
这就是消息队列的三大核心作用：解耦（上游不需要知道下游有谁）、异步（发完消息就返回，不等处理完）、削峰（突发流量先堆积在 MQ，下游按自己的能力消费）。
📐 消息队列通用模型
Producer
生产者
发送消息的应用
如：订单服务
→
Message Queue
消息队列（Broker）
存储 & 转发消息
独立的中间件服务
→
Consumer
消费者
处理消息的应用
如：库存/积分服务
Producer → 发消息(Message) → Broker(队列/主题) → Consumer 拉取/推送 → 处理业务逻辑
🎯 三大核心作用详解
1️⃣ 解耦（Decoupling）
订单服务发消息到 MQ，下游的库存服务、积分服务、短信服务各取所需。订单服务不需要知道下游有几个服务——新增一个数据分析服务，只需订阅对应的队列即可，订单服务代码完全不改。
代码对比：
❌ 没有 MQ：orderService.deductStock() → orderService.addPoints() → orderService.sendSMS()（耦合 3 个服务）
✅ 有了 MQ：orderService.send("order_created", orderInfo)（只管发消息，下游各自消费）
2️⃣ 异步（Async）
用户下单后，接口立即返回"下单成功"。扣库存、加积分、发短信这些耗时操作异步处理。用户感知的响应时间从同步的 700ms 降到 60ms。
耗时对比：
同步：下单(50ms) + 扣库存(200ms) + 加积分(150ms) + 发短信(300ms) = 700ms
异步：下单(50ms) + 发 MQ(10ms) = 60ms，其余操作异步完成
3️⃣ 削峰（Peak Clipping）
秒杀活动瞬间 10 万请求涌入，数据库扛不住。先把请求全部丢进 MQ，消费端按数据库能承受的速度（比如 1000/s）慢慢消费。MQ 起到了"蓄水池"的作用。
削峰效果：
没有削峰：10万请求 → 直接打数据库 → 数据库连接池耗尽 → 系统崩溃 💥
有了削峰：10万请求 → MQ 积压消息 → 消费端 1000/s 稳定消费 → 数据库安然无恙 ✅
🔍 消息队列的核心概念
| 概念 | 技术含义 | 说明 |
|---|
| Message | 传递的数据单元，包含 headers + body | 最小信息单位，如一条 JSON 格式的订单事件 |
| Producer | 发送消息的应用 | 消息的生产方，如订单服务、日志采集器 |
| Consumer | 接收处理消息的应用 | 消息的消费方，如库存服务、数据分析服务 |
| Broker | 消息队列服务器，负责存储转发 | 独立的中间件进程，如 RabbitMQ Server、Kafka Broker |
| Queue/Topic | 消息存放的地方 | RabbitMQ 叫 Queue，Kafka 叫 Topic |
| Exchange | RabbitMQ 特有：按规则把消息路由到不同队列 | 消息不直接进 Queue，先到 Exchange 做路由匹配 |
| Partition | Kafka 特有：一个 Topic 分成多个 Partition | 实现并行读写，Partition 数 = 最大并行消费数 |
| Consumer Group | Kafka 特有：同组消费者共同消费一个 Topic | 组内每个 Consumer 负责不同 Partition，互不重复 |
| ACK | 消费者确认消息已处理完毕 | 消费者发 ACK 后，Broker 才删除该消息 |
| Offset | Kafka 消息的位置编号 | 消费者记住自己消费到哪个 Offset，重启后从该位置继续 |
2. RabbitMQ 架构 — Exchange 路由模型
核心设计：RabbitMQ 的消息不是直接从 Producer 扔进 Queue 的。Producer 先把消息发给 Exchange（交换机），Exchange 根据 Routing Key + Binding 规则，决定把消息路由到哪个（或哪些）Queue。这个两层设计（Exchange → Queue）使得消息路由非常灵活——同一条消息可以根据规则分发到不同队列、做优先级处理、做按类型分发等。
📐 RabbitMQ 整体架构图
Binding = Exchange 和 Queue 之间的绑定规则（Binding Key）
Routing Key = 消息上标注的目标地址，Exchange 拿着它去匹配 Binding Key
🔍 核心组件详解
1. Connection（连接）
Producer / Consumer 与 RabbitMQ 之间的一条 TCP 长连接。所有通信都基于这个连接。
2. Channel（通道）
一条 Connection 上可以开多个 Channel（多路复用）。每个 Channel 是独立的轻量级虚拟连接，不用每次都建 TCP。
为什么需要 Channel？
一个应用可能要发很多种消息（订单、积分、短信），如果每种都建一个 TCP 连接，太浪费资源。
用 Channel：一个 TCP 连接 + 多个 Channel，每个 Channel 处理一种消息，效率高。
3. Exchange（交换机）
消息的第一站。Producer 的消息不是直接进 Queue，而是先到 Exchange。Exchange 根据类型 + Routing Key + Binding 规则决定把消息路由到哪个 Queue。
RabbitMQ 有 4 种 Exchange 类型（下一节详讲）。
4. Queue（队列）
真正存储消息的地方。FIFO（先进先出）。多个 Consumer 可以订阅同一个 Queue，但一条消息只会被其中一个 Consumer 消费（竞争消费）。
5. Binding（绑定）
Exchange 和 Queue 之间的关系。Binding Key 就是路由规则——"key=order" 的消息进 Queue1，"key=points" 的进 Queue2。
6. Virtual Host（虚拟主机）
RabbitMQ 里的"租户隔离"。每个 vhost 有自己的 Exchange、Queue、Binding、权限。不同业务线用不同 vhost，互不影响。默认 vhost 是 /。
📊 消息流转完整过程
1. Producer 创建消息
→
2. 选择 Exchange + Routing Key
→
3. Exchange 路由匹配
→
4. 投递到 Queue
→
5. Queue 存储消息
→
6. Consumer 拉取/推送
→
7. 处理业务逻辑
→
8. 发送 ACK
→
9. Queue 删除消息
完整链路：订单服务发消息（指定 Exchange 和 Routing Key）→ Exchange 根据 Binding 规则路由到目标 Queue → Consumer 从 Queue 取消息 → 执行业务逻辑（扣库存等）→ 发 ACK 确认 → RabbitMQ 才从 Queue 中删除该消息。如果 Consumer 处理失败发 NACK，消息会重新入队。
🖥️ RabbitMQ 管理界面
RabbitMQ Management — http://localhost:15672
Overview
📤 Messages ready: 1,234
📨 Messages unacked: 56
📊 Publish rate: 245/s
📥 Deliver rate: 230/s
Queues
| Name | Messages | Consumers | Rate |
|---|
| order_queue | 856 | 3 | 120/s |
| points_queue | 234 | 1 | 85/s |
| sms_queue | 144 | 2 | 60/s |
| dlx_order_queue | 12 | 0 | 0/s |
3. RabbitMQ 五种工作模式 — 5种路由策略
📦 模式 1：简单模式（Simple）— 一对一直送
最简单的模式：一个生产者、一个队列、一个消费者。没有 Exchange（用默认 Exchange），消息直接进 Queue。
适用场景：简单的点对点任务分发，如发一封邮件通知、执行一个后台批处理任务。
// Producer — 发送消息
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection conn = factory.newConnection();
Channel ch = conn.createChannel()) {
// 声明队列（不存在则自动创建）
ch.queueDeclare("order_queue", false, false, false, null);
// 发送消息
String msg = "{"orderId":1001,"product":"无线耳机","qty":2}";
ch.basicPublish("", "order_queue", null, msg.getBytes());
System.out.println(" [x] Sent '" + msg + "'");
}
// Consumer — 消费消息
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection conn = factory.newConnection();
Channel ch = conn.createChannel();
ch.queueDeclare("order_queue", false, false, false, null);
// 定义回调：收到消息时执行
DeliverCallback cb = (tag, delivery) -> {
String msg = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + msg + "'");
// 处理订单...
};
ch.basicConsume("order_queue", true, cb, tag -> {});
✅ 运行结果
Producer 输出: [x] Sent '{"orderId":1001,"product":"无线耳机","qty":2}'
Consumer 输出: [x] Received '{"orderId":1001,"product":"无线耳机","qty":2}'
📦 模式 2：工作队列模式（Work Queue）— 多消费者竞争消费
一条消息只会被一个 Consumer 消费（竞争消费），消息在多个消费者之间分配
一个生产者、一个队列、多个消费者。消息在消费者之间轮询分配（默认 Round-Robin）。适合任务分发场景——多个消费者分担处理同一个队列的消息，提高吞吐。
注意：能者多劳（Fair Dispatch）
默认是轮询（你一张我一张），但有个问题：Consumer A 处理快（100ms/条），Consumer B 处理慢（500ms/条）。如果平均分配，A 早就干完闲着了，B 还在忙。
解决：channel.basicQos(1) —— 一次只给一个消费者派 1 条消息，消费者处理完（ACK）后才给下一条。这样快的消费者会多处理，慢的少处理。
// Consumer — 开启公平分发
ch.basicQos(1); // 一次只处理 1 条消息，处理完 ACK 后才给下一条
DeliverCallback cb = (tag, delivery) -> {
String msg = new String(delivery.getBody(), "UTF-8");
System.out.println(" [Consumer-A] 处理: " + msg);
Thread.sleep(100); // 模拟处理耗时
ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // 手动 ACK
};
// 第二个参数 false = 手动 ACK（不自动确认）
ch.basicConsume("order_queue", false, cb, tag -> {});
✅ 运行结果（6 条消息，2 个消费者）
Consumer A: [Consumer-A] 处理: order-1（100ms 后 ACK）
Consumer B: [Consumer-B] 处理: order-2（500ms 后 ACK）
Consumer A: [Consumer-A] 处理: order-3（A 先 ACK，立即拿到下一条）
Consumer A: [Consumer-A] 处理: order-5（A 又先 ACK）
Consumer B: [Consumer-B] 处理: order-4
Consumer B: [Consumer-B] 处理: order-6
效果：A 处理了 4 条，B 处理了 2 条（能者多劳）
📦 模式 3：发布/订阅模式（Fanout）— 广播：所有队列都收到
→
Exchange
(fanout)
广播：不管 key
全部分发
→
同一条消息被复制到所有绑定的 Queue —— 每个 Queue 的 Consumer 都能收到完整消息
Exchange 类型为 fanout：不理会 Routing Key，消息被广播到所有绑定的 Queue。每个 Queue 都会收到一份消息的副本。
适用场景：订单创建后需要同时通知库存、积分、短信等多个下游服务。一条消息广播给所有绑定的队列。
// Producer — 广播消息
ch.exchangeDeclare("order_exchange", "fanout"); // 声明 fanout 类型交换机
String msg = "新订单: 无线耳机x2";
ch.basicPublish("order_exchange", "", null, msg.getBytes());
// Routing Key 为空，因为 fanout 根本不看 key
// Consumer 1（库存服务）— 每个消费者创建自己的临时队列
ch.exchangeDeclare("order_exchange", "fanout");
String queueName = ch.queueDeclare().getQueue(); // 随机队列名
ch.queueBind(queueName, "order_exchange", ""); // 绑定到交换机
ch.basicConsume(queueName, true, (tag, delivery) -> {
String msg = new String(delivery.getBody(), "UTF-8");
System.out.println(" [库存] 扣减库存: " + msg);
}, tag -> {});
✅ 运行结果（发送 1 条消息）
库存服务: [库存] 扣减库存: 新订单: 无线耳机x2
积分服务: [积分] 增加积分: 新订单: 无线耳机x2
短信服务: [短信] 发送通知: 新订单: 无线耳机x2
同一条消息被 3 个服务同时消费！
📦 模式 4：路由模式（Direct）— 按类型精确分发
→
Exchange
(direct)
按 key 精确匹配
→
Queue C
key=order
key=points
Routing Key = "order" → 进 Queue A 和 Queue C；Routing Key = "points" → 进 Queue B 和 Queue C
Exchange 类型为 direct：消息的 Routing Key 必须和 Binding Key 精确匹配才能进队列。
适用场景：按消息类型精确路由——"order" 类消息进订单队列、"points" 类消息进积分队列。比 fanout 更精细。
// Producer — 指定 Routing Key 发消息
ch.exchangeDeclare("direct_exchange", "direct");
ch.basicPublish("direct_exchange", "order", null, "订单消息".getBytes());
ch.basicPublish("direct_exchange", "points", null, "积分消息".getBytes());
ch.basicPublish("direct_exchange", "log", null, "日志消息".getBytes());
// Consumer — 只绑定自己关心的 key
ch.queueBind(queueName, "direct_exchange", "order"); // 只收 order
✅ 运行结果
库存服务(绑 order): 收到 "订单消息"
积分服务(绑 points): 收到 "积分消息"
日志服务(绑 log): 收到 "日志消息"
注意：key="log" 的消息只有日志服务收到，其他服务没收到（精确匹配）
📦 模式 5：主题模式（Topic）— 按通配符模糊匹配
→
Exchange
(topic)
按模式匹配
* = 一个词
# = 零或多个词
→
key="order.normal" → Queue A ✅ Queue B ❌ Queue C ✅
key="sms.urgent" → Queue A ❌ Queue B ✅ Queue C ✅
key="order.urgent" → Queue A ✅ Queue B ✅ Queue C ✅
Exchange 类型为 topic：Routing Key 可以用通配符匹配。
*（星号）：匹配恰好一个单词。如 order.* 匹配 order.create 但不匹配 order.create.vip#（井号）：匹配零或多个单词。如 order.# 匹配 order、order.create、order.create.vip
适用场景：需要按层级分类路由，如 "order.create"、"order.cancel.urgent" 等多级事件。
// Producer — 发送带层级 key 的消息
ch.exchangeDeclare("topic_exchange", "topic");
ch.basicPublish("topic_exchange", "order.create", null, "创建订单".getBytes());
ch.basicPublish("topic_exchange", "order.cancel.urgent", null, "加急取消".getBytes());
ch.basicPublish("topic_exchange", "sms.urgent", null, "短信加急".getBytes());
// Consumer A — 关心所有订单事件
ch.queueBind(queueName, "topic_exchange", "order.*");
// Consumer C — 全部消息（日志审计）
ch.queueBind(queueName, "topic_exchange", "#");
✅ 运行结果
消息 key="order.create"
→ Consumer A (order.*) ✅ | Consumer C (#) ✅
消息 key="order.cancel.urgent"
→ Consumer A (order.*) ❌ (* 只匹配一个词) | Consumer C (#) ✅
消息 key="sms.urgent"
→ Consumer A (order.*) ❌ | Consumer C (#) ✅
五种模式总结对比：
Simple = 1P→1Q→1C（简单直连）
Work Queue = 1P→1Q→多C（竞争消费）
Fanout = 1P→Exchange→所有Q（广播）
Direct = 1P→Exchange→精确匹配的Q（定向）
Topic = 1P→Exchange→模糊匹配的Q（模式匹配）
补充：还有一种 Headers 模式（根据消息头匹配），极少使用。
4. RabbitMQ 高级特性 — 怎么保证消息不丢？
🔒 4.1 消息确认（ACK）— 处理完才确认
RabbitMQ 默认是自动确认（autoAck=true）：消息一发给消费者就从队列删除。但如果消费者处理到一半崩了，消息就丢了。
手动确认（autoAck=false）：消费者处理完后主动发 ACK，RabbitMQ 才删除消息。如果消费者崩了没发 ACK，消息会重新入队。
核心机制：自动确认模式下，消息被取出就立即删除——如果消费者处理到一半崩溃，这条消息就永久丢失了。手动确认模式下，消息被取出后标记为 unacked，只有消费者处理完毕并发送 ACK 后才真正删除。如果消费者崩溃（TCP 断开），unacked 的消息会自动重新入队，被其他消费者处理。
// 手动 ACK 模式
Channel ch = conn.createChannel();
ch.basicQos(1); // 每次只发 1 条
DeliverCallback cb = (tag, delivery) -> {
String msg = new String(delivery.getBody(), "UTF-8");
try {
processOrder(msg); // 处理订单
// 处理成功 → ACK
ch.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
System.out.println(" [✓] ACK: " + msg);
} catch (Exception e) {
// 处理失败 → NACK，消息重新入队
ch.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
System.out.println(" [✗] NACK, 重新入队: " + msg);
}
};
// 第二个参数 false = 手动 ACK
ch.basicConsume("order_queue", false, cb, tag -> {});
⏰ 4.2 消息 TTL（Time To Live）— 消息有效期
消息可以设置过期时间，超过 TTL 的消息会被 RabbitMQ 自动删除（或进入死信队列）。
适用场景：订单 30 分钟未支付自动取消——订单消息设置 TTL=30min，过期后进入死信队列触发取消逻辑。
// 方式1：给队列设置 TTL（队列中所有消息统一过期时间）
Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 30000); // 30秒 = 30000ms
ch.queueDeclare("order_queue", false, false, false, args);
// 方式2：给单条消息设置 TTL（每条消息可以不同）
AMQP.BasicProperties props = new AMQP.BasicProperties()
.builder()
.expiration("60000") // 这条消息 60 秒后过期
.build();
ch.basicPublish("", "order_queue", props, msg.getBytes());
💀 4.3 死信队列（DLX）— 处理失败的异常消息
消息变成"死信"（Dead Letter）的三种情况：
- 消息被消费者 NACK 或 REJECT，且 requeue=false
- 消息超过 TTL 过期
- 队列达到最大长度，新消息被挤掉
死信不会被直接丢弃，而是被路由到一个死信交换机（DLX），再由死信交换机路由到死信队列，等待人工处理。
正常队列
order_queue
TTL=30s
满/被NACK/过期
→
→
→
为什么需要死信队列？在生产环境中，总有一些消息无法被正常消费（消费者报错、数据格式错误、依赖服务不可用等）。如果不做死信处理，这些消息会无限重试导致消息积压；如果直接丢弃，会造成数据丢失。死信队列是一种"异常消息兜底"机制——消费失败的消息被安全地转移到死信队列，等后续人工排查或补偿处理。
// 1. 创建死信交换机和死信队列
ch.exchangeDeclare("dlx_exchange", "direct");
ch.queueDeclare("dlx_order_queue", true, false, false, null);
ch.queueBind("dlx_order_queue", "dlx_exchange", "order");
// 2. 创建正常队列，绑定死信交换机
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx_exchange"); // 指定死信交换机
args.put("x-dead-letter-routing-key", "order"); // 死信路由 key
args.put("x-message-ttl", 30000); // 30秒 TTL
ch.queueDeclare("order_queue", true, false, false, args);
⏳ 4.4 延迟队列 — "30分钟后自动取消未支付订单"
RabbitMQ 没有原生的延迟队列，但可以通过 TTL + DLX 组合实现：
消息发到"等待队列"（设了 TTL）→ 消息过期 → 进入死信交换机 → 路由到"处理队列" → 消费者处理。
→
等待队列
TTL=30min
消息在这里等30分钟
没人消费
→
过期
→
→
实现原理："等待队列"设置了 30 分钟 TTL 但没有消费者。消息进入后无人消费，30 分钟后过期成为死信，被路由到死信交换机，再进入"处理队列"。消费者监听处理队列，收到消息后检查订单状态——如果已付款就跳过，未付款就执行取消逻辑。
// Spring Boot + RabbitMQ 延迟队列配置
@Configuration
public class DelayQueueConfig {
// 1. 死信交换机
@Bean
public DirectExchange orderDlxExchange() {
return new DirectExchange("order.dlx.exchange");
}
// 2. 处理队列（真正消费的队列）
@Bean
public Queue orderCancelQueue() {
return QueueBuilder.durable("order.cancel.queue").build();
}
// 3. 等待队列（TTL=30分钟，绑定死信交换机）
@Bean
public Queue orderWaitQueue() {
return QueueBuilder.durable("order.wait.queue")
.withArgument("x-dead-letter-exchange", "order.dlx.exchange")
.withArgument("x-dead-letter-routing-key", "cancel")
.withArgument("x-message-ttl", 1800000) // 30分钟 = 1800000ms
.build();
}
// 4. 绑定死信交换机 → 处理队列
@Bean
public Binding dlxBinding() {
return BindingBuilder.bind(orderCancelQueue())
.to(orderDlxExchange()).with("cancel");
}
}
// 5. 发送延迟消息
@Service
public class OrderService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void createOrder(Order order) {
// 保存订单到数据库...
orderMapper.insert(order);
// 发送延迟消息到等待队列
rabbitTemplate.convertAndSend("", "order.wait.queue",
JSON.toJSONString(order));
}
}
// 6. 消费延迟消息
@Component
public class OrderCancelConsumer {
@RabbitListener(queues = "order.cancel.queue")
public void handleCancel(String msg) {
Order order = JSON.parseObject(msg, Order.class);
if (order.getStatus() == OrderStatus.UNPAID) {
order.setStatus(OrderStatus.CANCELLED);
orderMapper.updateById(order);
log.info("订单{}超时未支付，已自动取消", order.getId());
}
}
}
✅ 效果
12:00:00 - 用户下单，发送延迟消息到 order.wait.queue
12:00:00 ~ 12:30:00 - 消息在等待队列中等待（无消费者）
12:30:00 - 消息过期 → 进入 DLX → 路由到 order.cancel.queue
12:30:00 - OrderCancelConsumer 消费：检查订单状态
12:30:00 - 订单1001状态=UNPAID → 自动取消 ✅
💾 4.5 持久化 — Broker 重启也不丢消息
RabbitMQ 默认数据在内存里，重启就没了。要保证消息不丢失，需要三重持久化：
⚠️ 必须三重持久化缺一不可！
1️⃣ Exchange 持久化（durable=true）
2️⃣ Queue 持久化（durable=true）
3️⃣ Message 持久化（deliveryMode=2）
只做 Queue 持久化不做 Message 持久化 = 队列还在但消息丢了，没意义。
// 1. Exchange 持久化
ch.exchangeDeclare("order_exchange", "direct", true); // 第3个参数 true = durable
// 2. Queue 持久化
ch.queueDeclare("order_queue", true, false, false, null); // 第2个参数 true = durable
// 3. Message 持久化
AMQP.BasicProperties props = new AMQP.BasicProperties()
.builder()
.deliveryMode(2) // 2 = 持久化，1 = 非持久化
.build();
ch.basicPublish("order_exchange", "order", props, msg.getBytes());
三层为什么要都做？
Exchange 持久化 = 交换机的元数据（名称、类型、Binding 规则）写入磁盘
Queue 持久化 = 队列的元数据写入磁盘
Message 持久化 = 消息体本身写入磁盘
缺任何一层，重启后对应的数据就不存在了——只有三层都持久化，Broker 重启后才能恢复完整的路由规则和消息内容。
🔄 4.6 生产者确认（Publisher Confirm）— 确认消息真的到了 Broker
生产者发消息后，怎么知道消息真的到了 RabbitMQ？用 Publisher Confirm 机制。
Producer 发消息
→
Broker 写入队列
→
Broker 回发 Confirm
→
Producer 收到确认
// 开启 Confirm 模式
ch.confirmSelect();
// 异步确认：监听 Confirm 回调
ch.addConfirmListener(new ConfirmListener() {
public void handleAck(long seq, boolean multiple) {
System.out.println(" [✓] 消息已确认: seq=" + seq);
}
public void handleNack(long seq, boolean multiple) {
System.out.println(" [✗] 消息被拒绝: seq=" + seq + "，需要重发！");
}
});
Spring Boot 配置：
spring.rabbitmq.publisher-confirm-type=correlated（异步回调）
spring.rabbitmq.publisher-returns=true（消息无法路由时回调）
5. RabbitMQ 集群 — 多节点高可用
为什么需要集群？单节点 RabbitMQ 存在单点故障风险——如果这台机器宕机，整个消息系统不可用。RabbitMQ 集群通过多节点部署 + 队列数据复制来保证高可用：任何一个节点故障，其他节点自动接管，消息不丢失。
📐 集群架构图
→
3 节点组成集群，元数据全节点同步，队列数据可配置镜像队列（HA Queue）
→
RabbitMQ 集群的两种高可用方案：
| 方案 | 原理 | 优点 | 缺点 | 版本 |
|---|
镜像队列
(HA Queue) |
每个队列在多个节点上有完整副本（master + mirrors） |
配置简单，真正的高可用 |
同步复制性能差，镜像太多浪费资源 |
3.x 以前 |
仲裁队列
(Quorum Queue) |
基于 Raft 协议的共识队列，多数节点写入成功即算成功 |
性能好、数据一致性强、推荐方案 |
不支持某些非持久化特性 |
3.8+ 推荐 |
// 仲裁队列声明（RabbitMQ 3.8+）
Map<String, Object> args = new HashMap<>();
args.put("x-queue-type", "quorum"); // 声明为仲裁队列
args.put("x-quorum-initial-group-size", 3); // 3 个节点
ch.queueDeclare("order_queue", true, false, false, args);
$ docker-compose up -d
Creating rabbit-node1 ... done
Creating rabbit-node2 ... done
Creating rabbit-node3 ... done
$ docker exec rabbit-node1 rabbitmqctl cluster_status
Cluster status of node rabbit@node1 ...
[{running_applications,[{rabbit,"RabbitMQ","3.12.0"}]}]
Disk Nodes: rabbit@node1, rabbit@node2, rabbit@node3
Running Nodes: rabbit@node1, rabbit@node2, rabbit@node3
✓ 集群 3 节点全部在线
6. Kafka 架构 — 日志模型与分区并行
核心设计：Kafka 与 RabbitMQ 有本质区别。RabbitMQ 是"消息代理"模型——消息被消费后即删除。Kafka 是"追加日志"模型——消息以日志文件形式持久化存储，消费后不删除（默认保留 7 天），可以被回放（重新消费）。
关键区别：RabbitMQ 是"邮递员模型"（送完就走），Kafka 是"日志模型"（追加写入，保留历史）。这使得 Kafka 特别适合大数据分析、日志收集、流处理等需要回放历史数据的场景。
📐 Kafka 整体架构图
Kafka Cluster（3 个 Broker）
→
Topic: order-events（3 个 Partition）
P0: [msg0] [msg3] [msg6] ... | P1: [msg1] [msg4] [msg7] ... | P2: [msg2] [msg5] [msg8] ...
→
Broker = Kafka 服务器节点 | Topic = 消息分类 | Partition = 并行分片 | Consumer Group = 消费者组
同一 Consumer Group 内：一个 Partition 只能被一个 Consumer 消费（不重复）
不同 Consumer Group：各自独立消费，都能收到全量数据
🔍 Kafka 核心概念详解
| 概念 | 技术含义 | 说明 |
|---|
| Broker | Kafka 服务器节点，默认端口 9092 | 一个集群有多个 Broker，数据分布在各节点上 |
| Topic | 消息的逻辑分类 | 如 order-events、user-behavior、access-log |
| Partition | 一个 Topic 分成多个 Partition | 分布在不同 Broker 上，实现并行读写。Partition 数 = 最大并行消费数 |
| Offset | 每条消息在 Partition 中的唯一编号（0, 1, 2, ...） | 单调递增，消费者记住 Offset 就知道消费到哪了 |
| Replica | 每个 Partition 有多个副本（Leader + Followers） | 保证高可用，Leader 挂了 Follower 顶上 |
| Leader/Follower | 读写都走 Leader，Follower 只负责同步 | Leader 挂了，ISR 中最新的 Follower 自动成为新 Leader |
| Consumer Group | 同组消费者共同消费一个 Topic 的所有 Partition | 组内互不重复，不同组各自独立消费全量数据 |
| ZooKeeper / KRaft | 管理集群元数据（哪些 Broker 活着、谁是 Leader） | 新版 Kafka 用 KRaft 替代 ZooKeeper |
📊 Partition 的数据结构
每个 Partition 就是一个追加写入的日志文件（Append-Only Log）：
Partition 0 — order-events
| Offset | Timestamp | Key | Value |
|---|
| 0 | 12:00:01 | order-1001 | {"product":"无线耳机","qty":2} |
| 1 | 12:00:03 | order-1002 | {"product":"蓝牙音箱","qty":1} |
| 2 | 12:00:05 | order-1003 | {"product":"充电宝","qty":3} |
| 3 | 12:00:08 | order-1004 | {"product":"数据线","qty":1} |
| Write Pointer → | 新消息追加到末尾，Offset 单调递增，永不修改 |
Partition 的三大特性：
1️⃣ 有序性：同一个 Partition 内消息严格按 Offset 顺序，先来先服务
2️⃣ 追加写入：只能在末尾添加，不能修改/删除历史消息（顺序 IO → 超级快！）
3️⃣ 分片并行：不同 Partition 可以并行读写，3 个 Partition = 3 倍吞吐
🖥️ 消息发到哪个 Partition？— 分区策略
| 策略 | 规则 | 适用场景 |
|---|
| 指定 Partition |
直接指定 partition 号 |
特殊需求 |
| 按 Key Hash |
hash(key) % partition数 |
最常用：保证同一个 key 的消息进同一个 Partition，从而保证有序 |
| Round Robin |
无 Key 时轮询分配 |
无序要求、均衡负载 |
| 自定义分区器 |
实现 Partitioner 接口 |
业务特殊需求（如 VIP 用户单独分区） |
// Java Producer — 指定 Key，自动按 Hash 分区
ProducerRecord<String, String> record = new ProducerRecord<>(
"order-events", // topic
"order-1001", // key → hash(key) % 3 决定进哪个 partition
"{"product":"无线耳机","qty":2}" // value
);
producer.send(record, (metadata, ex) -> {
System.out.println("分区=" + metadata.partition()
+ ", offset=" + metadata.offset());
});
// 输出: 分区=1, offset=42 (hash("order-1001") % 3 = 1)
7. Kafka 核心机制 — Consumer Group / 消息顺序 / Exactly-Once
👥 7.1 Consumer Group — 消费者如何分工
核心规则：同一个 Consumer Group 内，一个 Partition 只能被一个 Consumer 消费。不同 Group 之间互不影响。
Topic: order-events（3 个 Partition）
Consumer Group A（3个消费者）
✅ 完美分配，每人一个
Consumer Group B（1个消费者）
⚠️ 消费者不够，一人消费3个
Consumer Group C（5个消费者）
❌ C8、C9 闲置（消费者 > Partition）
重要规则：
• 消费者数 = Partition 数 → 完美分配，每人一个 ✅
• 消费者数 < Partition 数 → 部分消费者消费多个 Partition
• 消费者数 > Partition 数 → 多出的消费者闲置，浪费资源 ❌
• 所以 Kafka 的 Partition 数决定了最大并行消费数！
// Spring Boot Kafka Consumer 配置
@Configuration
public class KafkaConsumerConfig {
@Bean
public ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service-group"); // Group ID
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 从最早开始消费
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 手动提交 offset
return new DefaultKafkaConsumerFactory<>(props,
new StringDeserializer(), new StringDeserializer());
}
}
// 消费消息
@Component
public class OrderConsumer {
@KafkaListener(topics = "order-events", groupId = "order-service-group")
public void handleOrder(ConsumerRecord<String, String> record) {
log.info("收到消息: partition={}, offset={}, key={}, value={}",
record.partition(), record.offset(), record.key(), record.value());
// 处理订单...
}
}
📏 7.2 消息顺序保证
常见误解："Kafka 保证消息有序"——不完整！
Kafka 的顺序保证：
✅ 同一个 Partition 内：消息严格按写入顺序消费（Offset 单调递增）
❌ 不同 Partition 之间：不保证全局有序（P0 的消息可能比 P1 的消息先发后到）
解决方案：如果需要全局有序，只设 1 个 Partition（牺牲并行性）。
如果只需要同一个实体的操作有序（比如同一笔订单），把该订单的消息发到同一个 Partition（用 orderId 作为 Key，Kafka 会保证 hash(orderId) 一致的消息进同一个 Partition）。
为什么用 Key 可以保证有序？同一笔订单的"创建→支付→发货"这三个事件如果用 orderId 做 Key，它们会被 Hash 到同一个 Partition，在该 Partition 内按写入顺序排列，消费者按顺序消费就能保证业务操作的先后顺序正确。
🎯 7.3 Exactly-Once 语义 — 怎么保证"恰好一次"
消息系统有三种语义：
| 语义 | 含义 | 可能出现的问题 | 实现难度 |
|---|
| At Most Once | 最多一次（可能丢） | 消息发出后不等确认，如果传输中途失败则丢失 | 简单 |
| At Least Once | 至少一次（可能重复） | 消息确认前超时重发，导致消费者收到重复消息 | 中等 |
| Exactly Once | 恰好一次（不丢不重） | 需要 Producer 幂等 + Consumer 去重 + 事务保证 | 复杂 |
Kafka 通过 幂等 Producer + 事务 + Consumer 端去重 实现 Exactly-Once：
// Kafka Producer — 开启幂等（防止生产端重复）
props.put("enable.idempotence", "true"); // Kafka 自动去重
props.put("acks", "all"); // 所有副本确认后才返回成功
// Kafka 事务（Consumer → 处理 → Producer 端到端 Exactly-Once）
@Transactional
@KafkaListener(topics = "order-events")
public void processAndForward(ConsumerRecord<String, String> record) {
// 1. 处理消息
Order order = JSON.parseObject(record.value(), Order.class);
orderService.process(order);
// 2. 发送到下游 Topic
kafkaTemplate.send("order-processed", order.getId(), "PROCESSED");
// 3. 手动提交 Offset（和数据库事务绑定）
ack.acknowledge();
}
📝 7.4 Offset 管理
Consumer 怎么记住"我消费到哪了"？靠 Offset。
Offset 提交方式：
1️⃣ 自动提交（enable.auto.commit=true）：每隔 5s 自动提交。简单但可能丢消息或重复消费。
2️⃣ 手动同步提交（commitSync()）：处理完消息后同步提交。可靠但阻塞。
3️⃣ 手动异步提交（commitAsync()）：处理完消息后异步提交。性能好但可能提交失败。
4️⃣ 最佳实践：手动提交 + 和数据库事务绑定（消费+入库+提交 Offset 三者原子化）。
// 手动提交 Offset（推荐）
@KafkaListener(topics = "order-events")
public void handleOrder(ConsumerRecord<String, String> record,
Acknowledgment ack) {
try {
Order order = JSON.parseObject(record.value(), Order.class);
orderService.process(order); // 处理业务
ack.acknowledge(); // 手动提交 Offset
} catch (Exception e) {
log.error("处理失败，不提交 offset，消息会重新消费", e);
}
}
8. Kafka 高可用 — 副本机制与故障转移
📐 副本机制（Replica）
每个 Partition 有多个副本：1 个 Leader + N 个 Follower。
Topic: order-events, Partition 0（replication.factor=3）
Broker 1
Leader (P0)
负责读写
Offset: 0~999
←同步→
Broker 2
Follower (P0)
实时同步
Offset: 0~997
←同步→
Broker 3
Follower (P0)
实时同步
Offset: 0~996
Producer/Consumer 只和 Leader 交互。Follower 通过拉取（Fetch）保持同步。
如果 Broker 1 挂了，Broker 2（ISR 中最新的）自动成为新 Leader。
故障转移原理：所有读写都由 Leader 处理，Follower 只负责从 Leader 拉取数据保持同步。当 Leader 所在的 Broker 宕机时，Kafka 从 ISR（同步副本集）中选出数据最新的 Follower 作为新 Leader，继续提供读写服务。这个过程对 Producer 和 Consumer 透明——它们只需重连新 Leader 即可。
🔍 ISR（In-Sync Replicas）— 同步副本集
ISR = 和 Leader 保持同步的副本集合。Follower 如果落后太多（超过 replica.lag.time.max.ms），会被踢出 ISR。
ISR 关键配置：
• min.insync.replicas=2：至少 2 个副本写入成功才算成功（配合 acks=all）
• 如果 ISR 中只剩 1 个（只有 Leader），这条消息仍然可以写入，但降低了可靠性
• 最佳实践：replication.factor=3, min.insync.replicas=2, acks=all → 可以容忍 1 台机器故障
// Producer 配置 — 确保消息写入所有 ISR
props.put("acks", "all"); // 等待所有 ISR 副本确认（最可靠，最慢）
// acks=0: 不等确认（最快，可能丢）
// acks=1: 等 Leader 确认（中等）
// acks=all: 等所有 ISR 确认（最可靠）
// Topic 配置
// kafka-topics.sh --create --topic order-events \
// --partitions 3 --replication-factor 3 \
// --config min.insync.replicas=2
$ kafka-topics.sh --describe --topic order-events --bootstrap-server localhost:9092
Topic: order-events PartitionCount: 3 ReplicationFactor: 3
Topic: order-events Partition: 0 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: order-events Partition: 1 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1
Topic: order-events Partition: 2 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2
✓ 3 个 Partition，每个有 3 个副本，ISR 全部同步
⚡ Kafka 性能优化
| 优化方向 | 配置/方法 | 原理 |
|---|
| 批量发送 | batch.size=16384
linger.ms=5 | 攒一批消息再发，减少网络请求次数 |
| 压缩 | compression.type=lz4 | 消息压缩后传输，减少带宽 |
| 批量消费 | max.poll.records=500 | 一次拉 500 条批量处理 |
| 页缓存 | OS 级别自动 | Kafka 利用操作系统的 Page Cache，不自己管缓存 |
| 零拷贝 | OS 级别自动 | sendfile() 直接从文件到网卡，不经过用户空间 |
| 分区数 | Partition 数 = 期望并行度 | 更多 Partition = 更多并行，但也更多文件句柄 |
9. RocketMQ 概览 — 阿里出品的企业级消息队列
核心定位：RocketMQ 是阿里开源的消息队列，专为电商、金融等对消息可靠性要求极高的场景设计。
架构：Producer → NameServer（路由注册中心，告诉 Producer/Consumer 哪个 Broker 有哪个 Topic）→ Broker（消息存储服务器，Master 负责读写，Slave 负责备份）→ Consumer。
三大特色：事务消息（下订单和发消息要么同时成功要么同时失败）、延迟消息（原生支持！不用像 RabbitMQ 那样 TTL+DLX 拼凑）、消息过滤（按 Tag 过滤）。
📐 RocketMQ 架构图
NameServer：无状态的路由注册中心（类似注册中心），Producer/Consumer 通过它找到 Broker 地址
Broker：消息存储服务器，Master 负责读写，Slave 负责备份
Topic + Tag：Topic 是一级分类，Tag 是二级分类（如 Topic=ORDER, Tag=create/cancel）
🌟 RocketMQ 特色功能
1. 延迟消息（原生支持）
不用像 RabbitMQ 那样拼 TTL + DLX。RocketMQ 原生支持 18 个延迟级别（1s/5s/10s/30s/1m/.../2h）。
// RocketMQ 延迟消息
Message msg = new Message("ORDER_TOPIC", "create",
"{"orderId":1001}".getBytes());
msg.setDelayTimeLevel(16); // 第16级 = 30分钟
producer.send(msg);
2. 事务消息
本地事务（扣库存）和发消息（通知积分服务）原子化——要么都成功，要么都回滚。
// 事务消息
TransactionMQProducer producer = new TransactionMQProducer("order_group");
producer.setTransactionListener(new TransactionListener() {
// 1. 执行本地事务
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
try {
orderService.createOrder((Order) arg); // 本地事务：创建订单
return LocalTransactionState.COMMIT_MESSAGE; // 提交消息
} catch (Exception e) {
return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚消息
}
}
// 2. 回查本地事务状态（如果长时间没收到确认）
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
Order order = orderMapper.selectById(getOrderId(msg));
return order != null ? LocalTransactionState.COMMIT_MESSAGE
: LocalTransactionState.ROLLBACK_MESSAGE;
}
});
3. 消息过滤（Tag）
// Producer — 带 Tag 发送
Message msg = new Message("ORDER_TOPIC", "create", body);
// ↑ Tag
// Consumer — 按 Tag 过滤订阅
consumer.subscribe("ORDER_TOPIC", "create || cancel"); // 只收 create 和 cancel
4. 消息重试 & 死信
RocketMQ 消费失败自动重试：
• 默认重试 16 次：10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h
• 16 次仍然失败 → 进入死信 Topic（%DLQ%ConsumerGroup）
• 比 RabbitMQ 需要自己拼 DLX 方便多了
11. 消息可靠性全链路 — 怎么保证消息不丢？
全链路可靠性三段论：保证消息不丢需要覆盖消息生命周期的三个阶段：
① 生产端：Producer 发消息后必须确认 Broker 确实收到了（Publisher Confirm）
② Broker 端：消息到达后必须持久化到磁盘 + 多副本同步（防止单机故障）
③ 消费端：Consumer 必须处理完业务逻辑后才发 ACK（防止处理失败但消息已被删除）
① 生产端
Confirm / Retries
确保消息到达 MQ
→
→
① 生产端可靠性（RabbitMQ 为例）
// Spring Boot 配置 — 生产端可靠性
spring.rabbitmq.publisher-confirm-type=correlated // 异步 Confirm 回调
spring.rabbitmq.publisher-returns=true // 消息无法路由时回调
spring.rabbitmq.template.mandatory=true // 强制路由，无法路由时 ReturnCallback
// 自定义 Confirm 回调
@Component
public class RabbitConfirmCallback implements RabbitTemplate.ConfirmCallback {
@Override
public void confirm(CorrelationData data, boolean ack, String cause) {
if (ack) {
log.info("消息投递成功: {}", data.getId());
} else {
log.error("消息投递失败: {}, 原因: {}", data.getId(), cause);
// 重发或记录到数据库，后续补偿
}
}
}
② MQ 端可靠性
// RabbitMQ：Exchange + Queue + Message 三重持久化（见第 4.5 节）
// Kafka：acks=all + min.insync.replicas=2（见第 8 节）
③ 消费端可靠性
// Spring Boot — 手动 ACK 模式
spring.rabbitmq.listener.simple.acknowledge-mode=manual
spring.rabbitmq.listener.simple.prefetch=1 // 一次只发 1 条
@Component
public class OrderConsumer {
@RabbitListener(queues = "order_queue")
public void handleOrder(String msg, Channel ch,
@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try {
Order order = JSON.parseObject(msg, Order.class);
orderService.process(order); // 业务处理
ch.basicAck(tag, false); // 成功 → ACK
log.info("订单处理成功: {}", order.getId());
} catch (Exception e) {
ch.basicNack(tag, false, true); // 失败 → NACK + 重新入队
log.error("订单处理失败，重新入队", e);
}
}
}
✅ 三重保障下的完整链路
1. Producer 发消息 → Broker Confirm 确认 → 消息已到达 MQ ✅
2. MQ 持久化到磁盘 + 多副本同步 → 机器重启消息不丢 ✅
3. Consumer 取消息 → 处理业务 → 手动 ACK → MQ 才删除消息 ✅
4. Consumer 处理失败 → NACK → 消息重新入队 → 其他消费者处理 ✅
5. 多次失败 → 进入死信队列 → 人工介入 ✅
12. 幂等性方案 — 怎么保证"重复消费不出错"？
为什么需要幂等性？在分布式消息系统中，消息重复几乎不可避免——网络抖动导致 ACK 丢失、Consumer 重启后从旧 Offset 重新消费、Producer 超时重发等。如果没有幂等性保障，同一条消息被消费两次就会导致业务错误（扣两次库存、加两次积分）。幂等性的目标是：同一条消息消费一次和消费多次的结果完全一致。
方案 1：数据库唯一约束
// 利用数据库唯一索引去重
@Transactional
public void processOrder(Order order) {
// 1. 检查是否已处理（幂等表）
IdempotentRecord record = idempotentMapper.selectById(order.getId());
if (record != null) {
log.info("订单{}已处理，跳过", order.getId());
return; // 直接返回，不重复处理
}
// 2. 记录幂等（利用唯一索引，并发时只有一个成功）
idempotentMapper.insert(new IdempotentRecord(order.getId()));
// 3. 业务处理
stockService.deduct(order.getProductId(), order.getQty());
}
方案 2：Redis SETNX
// 利用 Redis 的 SETNX（Set if Not eXists）去重
public boolean isDuplicate(String orderId) {
Boolean success = redisTemplate.opsForValue()
.setIfAbsent("idempotent:" + orderId, "1",
24, TimeUnit.HOURS); // 24小时过期
return !success; // false = 第一次，true = 重复
}
public void processOrder(Order order) {
if (isDuplicate(order.getId())) {
log.info("重复消息，跳过");
return;
}
// 正常处理...
}
方案 3：状态机（乐观锁）
// 利用订单状态流转来保证幂等
@Update("UPDATE order SET status = 'PAID' WHERE id = #{id} AND status = 'UNPAID'")
int payOrder(String id);
// 返回 1 = 更新成功（第一次），返回 0 = 已经是 PAID（重复，跳过）
三种方案对比：
• 数据库唯一约束：最可靠，但有数据库开销。适合关键业务（支付）。
• Redis SETNX：高性能，但 Redis 宕机可能失效。适合高并发场景。
• 状态机：最优雅，但只适用于有状态流转的业务（订单状态、库存扣减）。
• 推荐组合：Redis SETNX 快速去重 + 数据库唯一约束兜底。
13. 消息积压处理 — 生产速率远超消费速率时怎么办？
什么是消息积压？当 Producer 发送消息的速率远超 Consumer 消费的速率时，消息会在 Broker 中大量堆积。如果积压持续增长，会导致：内存溢出、消息延迟增大、磁盘空间耗尽、消费者永远追不上。
常见原因：① 消费者处理逻辑变慢（如依赖的数据库慢查询）② 消费者实例宕机未及时发现 ③ 突发流量（秒杀、促销）④ 消费者代码 bug 导致大量 NACK 重试。
🔍 如何发现消息积压？
Grafana — RabbitMQ 监控面板
Messages Ready
52,431
↑ 比昨日 +4800%
Publish Rate
5,000/s
↑ 秒杀活动！
⚠️ 告警：队列 order_queue 消息积压 52,431 条，生产速率 5000/s 远超消费速率 230/s！
🛠️ 应急处理方案
方案 1：快速扩容消费者（最常用）
// Kafka：增加 Consumer 实例（注意不要超过 Partition 数！）
// 如果 Partition=3，最多 3 个 Consumer 有用
// 需要先扩 Partition：
# 将 Partition 从 3 扩到 10
kafka-topics.sh --alter --topic order-events --partitions 10 --bootstrap-server localhost:9092
// 然后部署更多 Consumer 实例（K8s 扩副本）
// kubectl scale deployment order-consumer --replicas=10
方案 2：临时转发 + 批量快速消费
// 写一个"快速消费"程序：只入库不处理业务，先快速消费掉积压消息
@Component
public class FastConsumer {
@RabbitListener(queues = "order_queue", concurrency = "20")
public void fastConsume(String msg, Channel ch, @Header long tag) {
// 快速写入数据库（批量 INSERT）
batchInsertMapper.insert(JSON.parseObject(msg, Order.class));
ch.basicAck(tag, false);
}
}
// 等积压消完，再用定时任务慢慢处理入库的订单
方案 3：消息 TTL + 死信兜底
给队列设 TTL（比如 1 小时），超时的消息自动进入死信队列。正常队列消费积压，死信队列后续人工处理。
这样可以保证正常队列不会无限膨胀导致内存溢出。
千万不要做：直接清空队列（消息全丢）！除非确认这些消息可以丢弃。正确做法是先扩容消费，积压的消息慢慢消化。
14. 最终一致性 — 分布式事务的"妥协方案"
强一致性 vs 最终一致性：
强一致性（不用 MQ）：订单服务调用库存、积分、短信服务，全部成功才返回"下单成功"。任何一个下游服务故障，整个下单失败。用户等待时间长，系统可用性差。
最终一致性（用 MQ）：订单服务发消息后立即返回"下单成功"。下游服务异步处理——积分可能晚 1 秒到账，短信可能晚 2 秒发送。用户感知"下单成功"了，数据"最终"会一致。这是互联网公司处理分布式事务的主流方案。
📐 最终一致性方案对比
| 方案 | 原理 | 优点 | 缺点 | 适合场景 |
|---|
| 本地消息表 |
业务操作和消息记录在同一个数据库事务中，后台线程扫描消息表发送到 MQ |
实现简单，不需要特殊 MQ 支持 |
消息表会膨胀，需要定时清理 |
最常用，适合大多数业务 |
事务消息
(RocketMQ) |
MQ 原生支持半消息机制，本地事务和消息提交原子化 |
不需要消息表，代码简洁 |
只有 RocketMQ 原生支持 |
用 RocketMQ 的项目 |
| TCC / Saga |
分布式事务框架（Seata），Try-Confirm-Cancel |
强一致性 |
复杂度高，性能差 |
金融级场景 |
📝 本地消息表方案详解（最实用！）
① 订单服务
1. 插入订单
2. 插入消息表
（同一事务）
→
→
→
// ===== 第一步：业务操作 + 写消息表（同一事务）=====
@Transactional
public void createOrder(Order order) {
// 1. 插入订单
orderMapper.insert(order);
// 2. 插入消息表（和订单在同一个事务里）
OutboxMessage msg = new OutboxMessage();
msg.setId(UUID.randomUUID().toString());
msg.setTopic("order-created");
msg.setKey(order.getId());
msg.setBody(JSON.toJSONString(order));
msg.setStatus("PENDING"); // 待发送
msg.setRetryCount(0);
msg.setCreateTime(new Date());
outboxMapper.insert(msg);
// 事务提交后：订单和消息要么都成功，要么都回滚
}
// ===== 第二步：定时任务扫描消息表，发送到 MQ =====
@Scheduled(fixedDelay = 5000) // 每 5 秒扫描一次
public void scanAndPublish() {
List<OutboxMessage> messages = outboxMapper.selectPending();
for (OutboxMessage msg : messages) {
try {
rabbitTemplate.convertAndSend(msg.getTopic(), msg.getKey(), msg.getBody());
msg.setStatus("SENT"); // 标记为已发送
outboxMapper.updateById(msg);
} catch (Exception e) {
msg.setRetryCount(msg.getRetryCount() + 1);
if (msg.getRetryCount() > 5) {
msg.setStatus("FAILED"); // 超过 5 次标记为失败
}
outboxMapper.updateById(msg);
}
}
}
// ===== 第三步：消费者正常消费 =====
@RabbitListener(queues = "points_queue")
public void addPoints(String msg) {
Order order = JSON.parseObject(msg, Order.class);
pointsService.addPoints(order.getUserId(), order.getAmount());
}
✅ 完整流程
1. 用户下单 → 订单入库 + 消息入表（同一事务，原子性 ✅）
2. 定时任务扫描 → 发送到 MQ → 标记已发送
3. 积分服务消费 → 加积分 → ACK
4. 如果 MQ 发送失败 → 重试（最多 5 次）→ 仍失败则标记 FAILED
5. FAILED 的消息 → 人工排查或补偿脚本处理
效果：即使 MQ 短暂不可用，消息也不会丢失（在消息表里）。等 MQ 恢复后定时任务自动补发。
🔍 最终一致性的补偿机制
即使有了本地消息表，也需要补偿机制兜底：
- 定时补偿：定时扫描长时间 SENT 但下游未确认的消息，重新发送
- 对账：每天凌晨对比订单表和积分表，找出不一致的数据
- 人工处理：FAILED 状态的消息，运营人员在后台手动处理
- 监控告警：MQ 积压 > 阈值、消费失败率 > 1% 时告警
总结：最终一致性 = 本地消息表 + 定时任务 + 重试 + 补偿 + 告警
不追求"实时一致"，而是保证"最终"（通常几秒到几分钟内）数据会一致。
这是互联网公司处理分布式事务的主流方案。