146｜消息队列核心价值：解耦、削峰、异步以及引入的复杂度

@Transactional
public Order placeOrder(OrderRequest request) {
    // 1. 扣减库存
    inventoryService.deduct(request.getSkuId(), request.getQuantity());
    // 2. 创建订单
    Order order = orderDao.insert(request);
    // 3. 发短信
    smsService.send(request.getPhone(), "下单成功！");
    // 4. 发邮件
    emailService.send(request.getEmail(), "Order Confirmed", template);
    // 5. 加积分
    pointsService.addPoints(request.getUserId(), 100);
    // 6. 写日志
    auditLogService.log("ORDER_PLACED", order.getId());
    // 7. 更新用户画像
    recommendationService.updateProfile(request.getUserId());
    return order;
}

短信服务挂了 → smsService.send() 抛异常 → 事务回滚 → 订单创建失败

订单服务 直接依赖：
├── 库存服务
├── 短信服务
├── 邮件服务
├── 积分服务
├── 日志服务
└── 推荐服务

@Transactional
public Order placeOrder(OrderRequest request) {
    inventoryService.deduct(request.getSkuId(), request.getQuantity());
    Order order = orderDao.insert(request);
    // 只发一条消息，剩下的不管
    mqProducer.send(new OrderPlacedEvent(order));
    return order;
}

生产者（Producer）──→  Broker（消息代理）──→  消费者（Consumer）
     发送消息           存储和路由消息           接收和处理消息

Producer → Queue → Consumer
           (一条消息只被一个消费者消费一次)

Producer → Topic → Subscriber A
                  → Subscriber B
                  → Subscriber C
           (一条消息被所有订阅者消费)

修改模块 A → 需要同时修改模块 B、C、D
模块 B 挂了 → 模块 A 也受影响
团队 B 开发慢 → 团队 A 被阻塞

没有 MQ：
订单服务 import 短信服务的 Client
订单服务 import 邮件服务的 Client
订单服务 import 积分服务的 Client
订单服务 import 日志服务的 Client
...（每加一个下游，订单服务就要加一个依赖）

有 MQ：
订单服务只 import 消息队列的 Client
（所有下游服务独立订阅消息，订单服务不需要知道它们的存在）

没有 MQ：
订单团队（5 人）做功能时，需要等短信团队（3 人）对齐接口
→ 短信团队排期到下个月 → 订单团队被阻塞

有 MQ：
订单团队定义好消息格式 {"orderId": "xxx", "userId": "yyy", "amount": 99.00}
→ 发到 order_topic
→ 短信团队自己订阅，自己解析，自己开发
→ 两个团队并行开发，互不阻塞

没有 MQ：
短信服务挂了 → 短信 SDK 调用失败 → 订单服务拿不到返回 → 创建订单失败

有 MQ：
短信服务挂了 → 短信消费者拉取消息失败 → 消息留在 Broker
→ 订单服务完全不受影响，继续正常创建订单
→ 短信服务恢复后，从 Broker 读取积压消息，继续处理

{
  "eventId": "uuid-xxx",
  "eventType": "ORDER_PLACED",
  "timestamp": "2026-09-04T10:00:00Z",
  "data": {
    "orderId": "12345",
    "userId": "67890",
    "amount": 99.00,
    "skuId": "SKU-001",
    "quantity": 2
  }
}

秒杀场景：
  平时：100 QPS
  秒杀开始：10000 QPS（100 倍）
  秒杀结束：100 QPS

如果不做削峰：
  数据库需要按 10000 QPS 配置 → 平时 99% 的资源闲置
  按 100 QPS 配置 → 秒杀时直接打挂

          ┌─────────────────────────┐
          │      10000 QPS 流量      │
          │  ↓  ↓  ↓  ↓  ↓  ↓  ↓   │
          │  ┌─────────────────┐    │
          │  │   消息队列(蓄水池) │    │
          │  │  消息堆积 100 万条  │    │
          │  └────────┬────────┘    │
          │           ↓  ↓  ↓       │
          │      1000 QPS 消费      │
          └─────────────────────────┘

Kafka：磁盘可以存 TB 级别的消息（消息持久化到磁盘，可配置保留时间）
RocketMQ：磁盘存储，同样支持海量消息堆积
RabbitMQ：默认内存存储，大量堆积会耗尽内存，需要配置 Lazy Queue

消费速率 = 消费者实例数 × 单实例并发度 / 单条消息处理时间

例如：
- 10 个消费者实例
- 每个实例 20 个线程
- 每条消息处理 50ms
→ 消费速率 = 10 × 20 / 0.05 = 4000 条/秒

积压时间 = 堆积消息量 / (消费速率 - 生产速率)

例如：
- 秒杀 10 分钟，每秒产生 10000 条，共 600 万条
- 消费速率 1000 条/秒，秒杀结束后生产速率降为 100 条/秒
- 积压时间 = 600 万 / (1000 - 100) ≈ 6667 秒 ≈ 1.85 小时

用户请求 → 订单服务 → 库存服务 → 返回结果 → 订单服务 → 短信服务 → 返回结果 → ... → 响应用户

用户请求 → 订单服务（核心操作）→ 立即响应用户
                ↓
          发消息到 MQ
                ↓
      短信服务（异步处理）
      邮件服务（异步处理）
      积分服务（异步处理）

OrderCreated → PaymentRequested → PaymentConfirmed → InventoryDeducted → ShipmentPrepared

核心流程（必须同步完成）：
├── 库存扣减
└── 订单创建

非核心流程（可以异步处理）：
├── 发送短信
├── 发送邮件
├── 增加积分
├── 写入日志
└── 更新用户画像

核心流程成功 = 订单创建成功 = 返回用户"下单成功"
非核心流程失败 = 消息重试，或进入死信队列，人工处理

订单状态：已下单 → 已支付 → 已发货 → 已完成
通知状态：短信已发送 ✓ | 邮件已发送 ✓

            ┌─────────────────┐
            │   引入消息队列    │
            └────────┬────────┘
                     │
    ┌────────────────┼────────────────┐
    ↓                ↓                ↓
 消息丢失          消息重复          消息顺序
    ↓                ↓                ↓
 需要确认机制      需要幂等设计      需要顺序保证
    ↓                ↓                ↓
    └────────────────┼────────────────┘
                     ↓
              系统复杂度 ↑↑↑
              运维成本 ↑↑↑
              排查难度 ↑↑↑

生产阶段：Producer → Broker（网络抖动，消息没发出去）
存储阶段：Broker 内存（Broker 宕机，内存消息丢失）
消费阶段：Broker → Consumer（消费者挂了，消息没确认就被删了）

场景 1：Producer 发送消息，Broker 收到了，但 ACK 网络丢包
       → Producer 重发 → 同一条消息在 Broker 中出现两次

场景 2：Consumer 消费消息，处理完了，但在 ACK 之前宕机
       → Consumer 重启 → 重新消费同一条消息

// 幂等消费：利用数据库唯一约束
@Transactional
public void consume(OrderMessage msg) {
    // 以消息 ID 作为幂等键
    if (idempotentDao.exists(msg.getMessageId())) {
        return; // 已经处理过，跳过
    }
    // 处理业务逻辑
    processOrder(msg);
    // 记录幂等键
    idempotentDao.insert(msg.getMessageId());
}

Producer 按顺序发送：A → B → C
Partition 1：A → C（B 可能去了其他分区）
Consumer 并行消费：先拿到 C，后拿到 A

场景：创建订单（写数据库） + 发消息（写 MQ）

问题：数据库写入成功，但消息发送失败 → 数据不一致
问题：消息发送成功，但数据库写入失败 → 数据不一致

需要部署和维护的组件：
├── 消息队列集群（至少 3 个节点）
├── 监控系统（Lag、吞吐量、磁盘使用率）
├── 死信队列处理
├── 消息积压应急脚本
└── 消息格式版本管理

用户注册 → 写入数据库 → 发送 UserRegistered 事件
                            ↓
               ┌────────────┼────────────┐
               ↓            ↓            ↓
          发送欢迎邮件   初始化账户    赠送新人礼包

MySQL 订单表 → Binlog → Canal → MQ → ES 索引更新 → 搜索服务
                                  → Redis 缓存更新 → 查询服务
                                  → 数据仓库 → 报表服务

用户行为日志 → Kafka → Flink 实时计算 → 实时大屏
                               → 异常检测 → 告警
                               → 用户画像 → 推荐系统

Command（写）：POST /orders → OrderCommand → MQ → OrderCommandHandler → DB
Query（读）：  GET /orders/123 → OrderQueryHandler → ReadModel(ES/Redis)

账户余额 = 初始余额 + 所有充值事件 - 所有扣款事件

如果被调用方返回结果对调用方后续逻辑至关重要 → 同步调用更合适
如果只是"通知"性质的 → 可以考虑异步

如果消息丢失会导致严重的业务问题（如资金丢失） → 需要额外的保障机制
如果消息丢失只是"少发了一条通知" → 可以接受

如果团队只有 3 个人，已经负责了 5 个系统 → 引入 MQ 是给自己找麻烦
如果公司有专门的中间件团队 → 可以考虑

需要异步处理？
├── 否 → 不要用 MQ
└── 是 → 需要解耦多个系统？
    ├── 否 → 考虑线程池 / Spring Event
    └── 是 → 需要削峰填谷？
        ├── 否 → 考虑简单的 MQ（如 Redis List）
        └── 是 → 需要高可靠？
            ├── 否 → 考虑轻量 MQ（如 Redis Stream）
            └── 是 → 使用专业 MQ（Kafka / RocketMQ / RabbitMQ）

当前架构：
- 订单服务（单体 Spring Boot 应用）
- 每日订单量：10 万
- 峰值 QPS：500
- 下单接口平均响应时间：200ms

业务需求：
- 即将上线"双十一大促"，预计峰值 QPS 5000
- 需要新增"下单后推荐相似商品"功能
- 需要对接第三方风控系统

// 改造前：同步处理所有逻辑
@Transactional
public Order placeOrder(OrderRequest req) {
    Order order = orderDao.insert(req);          // 5ms
    inventoryService.deduct(req);                 // 10ms
    couponService.use(req.getCouponId());         // 15ms
    smsService.send(req.getPhone());              // 80ms
    emailService.send(req.getEmail());             // 80ms
    auditLog.write("ORDER_PLACED");               // 10ms
    return order;  // 总耗时 ≈ 200ms
}

// 改造后：核心同步，非核心异步
@Transactional
public Order placeOrder(OrderRequest req) {
    Order order = orderDao.insert(req);          // 5ms
    inventoryService.deduct(req);                 // 10ms
    couponService.use(req.getCouponId());         // 15ms
    // 发消息，以下操作异步处理
    mqProducer.send(new OrderPlacedEvent(order));  // 2ms
    return order;  // 总耗时 ≈ 32ms
}

用户请求 → 网关限流（5000 QPS）→ MQ → 订单服务（1000 QPS 稳定消费）→ 数据库

// 推荐服务：完全独立部署，不依赖订单服务
@RocketMQMessageListener(topic = "order_topic", consumerGroup = "recommendation")
public class RecommendationConsumer implements RocketMQListener<OrderPlacedEvent> {
    @Override
    public void onMessage(OrderPlacedEvent event) {
        // 根据订单信息更新用户画像，推荐相似商品
        recommendationService.updateProfile(event.getUserId(), event.getItems());
    }
}

// 风控服务：同样独立部署
@RocketMQMessageListener(topic = "order_topic", consumerGroup = "risk_control")
public class RiskControlConsumer implements RocketMQListener<OrderPlacedEvent> {
    @Override
    public void onMessage(OrderPlacedEvent event) {
        // 异步风控检查，高风险订单标记为"待审核"
        riskControlService.check(event.getOrderId());
    }
}

                      ┌──────────────────┐
                      │   网关（限流）     │
                      └────────┬─────────┘
                               ↓ 5000 QPS
                      ┌──────────────────┐
                      │   RocketMQ        │
                      │   order_topic     │
                      └────────┬─────────┘
                               ↓ 1000 QPS
                      ┌──────────────────┐
                      │   订单服务        │
                      │   (核心流程)      │
                      └────────┬─────────┘
                               ↓ 发送 OrderPlacedEvent
                      ┌──────────────────┐
                      │   RocketMQ        │
                      │   order_event_topic│
                      └──┬──────┬──────┬─┘
                         ↓      ↓      ↓
                   短信服务  推荐服务  风控服务

阶段一：Producer 注册新 Schema（FULL 兼容模式）
→ Registry 检查通过
阶段二：升级所有 Consumer 到能解析新 Schema 的版本
→ 此时新字段对 Consumer 可见，但 Producer 还没开始发新字段
阶段三：升级 Producer 开始发送新字段
→ Consumer 已经准备好了，平滑过渡
阶段四：确认所有 Consumer 不再需要老字段，从 Schema 中删除

方案：增加 Consumer 实例数量
原理：RocketMQ 中，同一个 Consumer Group 下，一个 MessageQueue 
      同一时间只能被一个 Consumer 消费。但如果有 16 个 MessageQueue，
      可以部署 16 个 Consumer 实例并行消费。
限制：Consumer 数 ≤ MessageQueue 数（多了也没用）

秒杀订单消息 → 拆分为：
  Priority 1（高优）：订单状态变更通知（用户需要立即看到"已支付"）
  Priority 2（中优）：短信、邮件通知（用户可接受 1-2 分钟延迟）
  Priority 3（低优）：积分、推荐画像更新（10 分钟延迟也可接受）

优化前：
  SMS 消费者：拉消息 → 查数据库校验订单 → 调短信 API → 写日志 → 确认消费
  耗时：200ms/条

优化后：
  - 批量拉取消息（一次拉 100 条，而非逐条）
  - 短信 API 改为异步批量发送（一次请求发 100 条短信）
  - 日志改为异步写入
  - 去掉不必要的 DB 校验（消息在发送时已验证）
  耗时：2ms/条（100 倍提升）

如果积压实在太深（如超过 100 万条）：
  → 低优消息（积分、推荐）直接跳过，避免占消费资源
  → 跳过的消息通过"离线批处理"（如每小时跑一次批处理任务）补数据
  → 效果：让高优消息快速出队，低优消息用批处理保底

// RocketMQ 事务消息
TransactionMQProducer producer = new TransactionMQProducer("order_group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 1. 执行本地事务：创建订单
            orderDao.insert(order);
            return LocalTransactionState.COMMIT_MESSAGE;  // 提交消息
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚消息
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 2. 事务回查：Broker 没收到确认时，主动问 Producer 事务状态
        String orderId = msg.getKeys();
        Order order = orderDao.findById(orderId);
        return order != null ? COMMIT_MESSAGE : ROLLBACK_MESSAGE;
    }
});

┌──────────────┐     ┌──────────────────┐     ┌──────────────┐
│  订单服务     │────→│  本地消息表       │────→│  定时任务     │
│  (本地事务)   │     │  (同库同事务)     │     │  (扫描发送)   │
└──────────────┘     └──────────────────┘     └──────┬───────┘
                                                     │
                                              ┌──────▼───────┐
                                              │  RocketMQ     │
                                              └──────────────┘

@Transactional
public void createOrder(Order order) {
    // 同一个本地事务中：
    orderDao.insert(order);                      // 1. 创建订单
    messageDao.insert(new OutboxMessage(         // 2. 写入本地消息表
        "order_topic", order.toEventJson(), "PENDING"
    ));
}

// 定时任务：每秒扫描 PENDING 消息
@Scheduled(fixedDelay = 1000)
public void sendPendingMessages() {
    List<OutboxMessage> messages = messageDao.findByStatus("PENDING");
    for (OutboxMessage msg : messages) {
        try {
            mqProducer.send(msg.toMQMessage());
            messageDao.updateStatus(msg.getId(), "SENT");  // 标记已发送
        } catch (Exception e) {
            // 发送失败，下次定时任务重试
        }
    }
}

定时任务（如每小时）：
  扫描今天创建的订单 → 查消息消费记录 → 发现缺少消息的订单 →
  补发消息或触发告警人工介入

@RocketMQMessageListener(topic = "order_topic", consumerGroup = "inventory")
public class InventoryConsumer implements RocketMQListener<OrderPlacedEvent> {
    
    @Autowired
    private InventoryService inventoryService;
    @Autowired
    private DedupService dedupService;
    
    @Override
    public void onMessage(OrderPlacedEvent event) {
        String dedupKey = "dedup:inventory:" + event.getOrderId();
        
        // 1. Redis 去重：SET NX（原子操作，只有不存在时才设置成功）
        boolean isNew = dedupService.tryLock(dedupKey, 3600); // 1 小时过期
        if (!isNew) {
            log.info("重复消息，跳过: orderId={}", event.getOrderId());
            return;  // 已处理过，直接返回
        }
        
        try {
            // 2. 执行业务逻辑：扣减库存
            inventoryService.deduct(event.getSkuId(), event.getQuantity());
        } catch (InsufficientStockException e) {
            // 库存不足可能是"已经扣减过"导致的
            // 查询当前库存状态，确认是否为重复消费
            int currentStock = inventoryService.getStock(event.getSkuId());
            int expectedAfterDeduct = currentStock + event.getQuantity();
            // 如果当前库存 + 本次扣减量 = 上次扣减前的库存，说明是重复消费
            log.warn("库存不足，可能是重复消费: orderId={}, currentStock={}", 
                     event.getOrderId(), currentStock);
            // 已经扣减过了，算成功
        }
    }
}

@RocketMQMessageListener(topic = "order_topic", consumerGroup = "inventory")
public class InventoryConsumer implements RocketMQListener<OrderPlacedEvent> {
    
    @Override
    public void onMessage(OrderPlacedEvent event) {
        try {
            // 利用数据库唯一约束实现幂等
            // 插入一条"消费记录"，如果已存在则抛 DuplicateKeyException
            consumeRecordDao.insert(new ConsumeRecord(
                event.getOrderId(), "inventory_consumer", "PROCESSING"
            ));
        } catch (DuplicateKeyException e) {
            log.info("重复消费，跳过: orderId={}", event.getOrderId());
            return;
        }
        
        // 执行业务
        inventoryService.deduct(event.getSkuId(), event.getQuantity());
        
        // 更新消费记录状态
        consumeRecordDao.updateStatus(event.getOrderId(), "DONE");
    }
}

需要 → 同步 RPC
  例：用户登录 → 必须立即返回"登录成功"或"密码错误"
  例：支付确认 → 必须立即返回"支付成功"或"余额不足"

不需要 → 消息队列
  例：下单后发短信 → 用户不需要等短信发完才看到"下单成功"
  例：用户行为日志 → 异步记录即可

依赖 → 同步 RPC
  例：创建订单前需要查库存 → 库存不足则拒绝下单
  例：下单前需要风控校验 → 风控拒绝则订单不创建

不依赖 → 消息队列
  例：创建订单后通知推荐系统 → 推荐系统是否成功不影响订单

不能 → 同步 RPC（或至少需要同步确认）
  例：支付服务故障 → 不能接受订单（钱收不到）

能 → 消息队列
  例：短信服务故障 → 订单仍然可以创建

用户请求
  │
  ▼
网关
  │
  ▼
订单服务（核心流程）
  ├── 同步调用：库存服务（必须拿到结果）
  ├── 同步调用：风控服务（必须拿到结果）
  ├── 同步调用：支付服务（必须拿到结果）
  │
  └── 异步发送：OrderPlacedEvent
       ├──→ 短信服务（允许延迟）
       ├──→ 邮件服务（允许延迟）
       ├──→ 积分服务（允许延迟）
       └──→ 推荐服务（允许延迟）