🔗 消息队列因果链 — 因为…所以…

📊 总览 ① 同步→MQ ② 消息丢失 ③ 消息重复→幂等 ④ 顺序消费 ⑤ 消费失败→死信 ⑥ 延迟需求→延迟队列 ⑦ 积压应急 🎯 面试点

📊 因果链总览图

❓ 同步调用
阻塞/耦合 → 💡 引入MQ
异步解耦 → ⚡ 消息可能丢失 → 💡 确认机制+持久化

⚡ 消息重复消费 → 💡 幂等设计 → ⚡ 消息顺序要求 → 💡 顺序消费方案

⚡ 消费失败 → 💡 死信队列 → ⚡ 延迟需求 → 💡 延迟队列

⚡ 流量突增→积压 → 💡 积压应急方案

类比：快递中转站
同步 = 你亲自去仓库取货 → MQ = 快递中转站（你下单，快递小哥帮你送）
消息丢失 = 快递丢了 → 幂等 = 同一个包裹收两次只算一次 → 顺序 = 按发货顺序收 → 死信 = 地址不对退回来 → 延迟 = 预约明天送 → 积压 = 双11爆仓

① 因为：同步调用阻塞/耦合 → 所以引入MQ

❓ 问题：下单时同步调用库存+支付+通知，一个慢全慢

// ❌ 同步调用：3个服务串行执行
public void createOrder(OrderDTO dto) {
    orderDao.insert(dto);                    // 50ms
    inventoryService.deduct(dto);            // 200ms ← 慢！
    paymentService.pay(dto);                 // 300ms ← 更慢！
    notificationService.sendConfirm(dto);    // 100ms ← 不重要也等
    // 总耗时：650ms，用户等了0.65秒
    // 如果支付服务挂了 → 用户直接看到500错误
}

💡 所以：引入MQ——异步解耦

下单服务 → MQ
（消息队列） → 库存服务

发完消息就返回异步消费支付服务

通知服务

// ✅ MQ异步：下单服务只管发消息
@Service
public class OrderService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void createOrder(OrderDTO dto) {
        orderDao.insert(dto);  // 50ms
        // 发消息，不用等下游处理完
        rocketMQTemplate.convertAndSend("order-created", dto);
        // 总耗时：50ms + 发消息5ms = 55ms
        // 下游服务异步消费，不影响用户
    }
}

// 库存服务消费消息
@RocketMQMessageListener(topic = "order-created", consumerGroup = "inventory-group")
public class InventoryConsumer implements RocketMQListener<OrderDTO> {
    @Override
    public void onMessage(OrderDTO dto) {
        inventoryService.deduct(dto);
    }
}

⚡ 但是：MQ可能丢消息！

发送方发完消息 → MQ还没持久化就宕机 → 消息丢失！
消费方取了消息 → 处理到一半挂了 → 消息丢失！

② 因为：消息可能丢失 → 所以确认机制+持久化

❓ 问题：消息在三个环节可能丢

1. 生产端：发送到MQ时网络抖动 → 消息没到MQ
2. MQ端：MQ收到但没持久化就宕机 → 消息在内存中丢失
3. 消费端：消费者取了消息但处理失败 → 消息没了

💡 所以：生产端确认 + MQ持久化 + 消费端手动ACK

// 1. 生产端：确认机制
// RabbitMQ: Confirm模式
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
    if (!ack) {
        log.error("消息发送失败: {}", cause);
        // 重发或记录到DB待补偿
    }
});

// 2. MQ端：消息持久化
// RabbitMQ: 队列持久化 + 消息持久化
@Bean
public Queue orderQueue() {
    return QueueBuilder.durable("order-queue").build(); // durable=持久化
}

// 3. 消费端：手动ACK
@RabbitListener(queues = "order-queue")
public void onMessage(OrderDTO dto, Channel channel,
                     @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
    try {
        orderService.process(dto);
        channel.basicAck(tag, false); // ✅ 手动确认：处理完了才ACK
    } catch (Exception e) {
        channel.basicNack(tag, false, true); // ❌ 处理失败：重新入队
    }
}

⚡ 但是：重试/网络抖动 → 消息重复消费

消费端处理成功了，但ACK时网络断了 → MQ以为没消费 → 重新投递 → 同一条消息被消费两次！

③ 因为：消息重复消费 → 所以幂等设计

❓ 问题：同一个订单扣了两次库存

消息1：订单A扣库存 → 库存-1 ✅

消息1重试：订单A扣库存 → 库存又-1 ❌ 重复扣了！

💡 所以：幂等设计——同一操作执行多次结果相同

// 幂等方案1：唯一ID + 去重表
@Service
public class IdempotentConsumer {
    @Autowired
    private DeduplicationMapper dedupMapper;

    public void consume(OrderDTO dto) {
        String msgId = dto.getOrderId() + "_" + dto.getType();
        // 幂等Key：订单ID + 操作类型

        try {
            // 插入去重表（唯一索引）
            dedupMapper.insert(msgId);
        } catch (DuplicateKeyException e) {
            log.info("消息已消费过: {}", msgId);
            return; // ✅ 已消费过，直接返回
        }

        // 真正的业务逻辑
        inventoryService.deduct(dto);
    }
}

// 幂等方案2：状态机
// 因为订单状态只能单向流转：待支付→已支付→已发货
// 所以用乐观锁：
@Update("UPDATE orders SET status='PAID' WHERE id=#{id} AND status='UNPAID'")
int payOrder(Long id);
// 返回0 = 状态已变更，说明已处理过 → 幂等

类比：不幂等 = 同一订单扣两次钱
你买了一个iPhone，微信扣了8999，但因为网络抖动又扣了8999 → 买了两个？
幂等 = 收银台认单号，同一个单号不管付几次只扣一次

④ 因为：业务要求消息有序 → 所以顺序消费

❓ 问题：创建→支付→发货，如果消费乱序怎么办？

3条消息进入3个队列，消费者并发消费 → 可能先消费"发货"再消费"创建" → 逻辑错误

💡 所以：同一业务ID的消息发到同一个队列

// RocketMQ 顺序消息
// 因为：同一个订单的消息必须按顺序消费
// 所以：用orderId作为sharding key → 同一订单的消息到同一队列

// 发送顺序消息
rocketMQTemplate.setMessageQueueSelector((mqs, msg, arg) -> {
    Long orderId = (Long) arg;
    // 用订单ID选队列：同一订单的消息到同一队列
    int index = (int) (orderId % mqs.size());
    return mqs.get(index);
});

// 消费端：单线程消费（保证顺序）
@RocketMQMessageListener(
    topic = "order-events",
    consumerGroup = "order-sequence-group",
    consumeMode = ConsumeMode.ORDERLY  // ✅ 顺序消费
)

类比：翻书 vs 查字典
普通消费 = 查字典，随便翻哪页都行
顺序消费 = 翻小说，必须从第1页到第2页到第3页，不能跳

⑤ 因为：消费失败重试还是失败 → 所以死信队列

❓ 问题：消息重试3次还是失败，怎么办？

一直重试？其他消息都堵在后面了。直接丢弃？那是业务数据，丢了就出大事。

💡 所以：死信队列——重试失败的消息进入特殊队列等待人工处理

// RabbitMQ 死信队列配置
@Bean
public Queue orderQueue() {
    return QueueBuilder.durable("order-queue")
        .withArgument("x-dead-letter-exchange", "order-dlx")      // 死信交换机
        .withArgument("x-dead-letter-routing-key", "order-dlq") // 死信路由
        .withArgument("x-message-ttl", 60000)                      // 消息TTL
        .build();
}

// 死信消费者：人工处理或自动补偿
@RabbitListener(queues = "order-dlq")
public void handleDeadLetter(OrderDTO dto) {
    log.error("消息进入死信: {}", dto);
    // 1. 记录到DB，人工后续处理
    // 2. 发告警通知运维
    // 3. 自动补偿逻辑
}

类比：无法投递的退信
快递送了3次都没人签收 → 放到邮局的"退信区" → 等收件人来取

⑥ 因为：有延迟需求 → 所以延迟队列

❓ 问题：下单30分钟未支付 → 自动取消

💡 所以：延迟队列——消息延迟指定时间后才能被消费

// RabbitMQ: TTL + 死信 = 延迟队列
// 因为：消息需要延迟N秒后消费
// 所以：消息设置TTL → 过期后进入死信队列 → 消费死信队列

@Bean
public Queue orderDelayQueue() {
    return QueueBuilder.durable("order-delay-queue")
        .withArgument("x-dead-letter-exchange", "order-cancel-exchange")
        .withArgument("x-message-ttl", 1800000) // 30分钟
        .build();
}

// RocketMQ: 原生延迟消息（更简单）
Message msg = MessageBuilder.withPayload(dto)
    .setDelayLevel(16) // 16级 = 30分钟
    .build();
rocketMQTemplate.syncSend("order-timeout", msg);

⑦ 因为：流量突增 → 所以消息积压应急

❓ 问题：MQ积压100万条消息，消费速度跟不上

💡 所以：紧急扩容消费者 + 跳过非核心消息

// 积压应急方案：
// 1. 快速扩容消费者数量（从5个扩到50个）
// 2. 跳过非核心消息（日志类直接丢弃）
// 3. 临时增加消费线程数
// 4. 如果来不及：新建临时Topic，把积压消息快速转发到新Topic
//    由新的消费者集群并行消费

// 预防措施：
// - 设置队列最大长度（防内存溢出）
// - 监控消费延迟（lag > 10000 就告警）
// - 消费者做好优雅停机（不丢正在处理的消息）

🎯 面试点映射

面试题	因为…所以…
为什么用MQ	因为同步调用阻塞+耦合 → 所以MQ异步解耦+削峰
如何保证不丢消息	因为3个环节都可能丢 → 生产端确认+MQ持久化+消费端手动ACK
如何保证幂等	因为网络抖动导致重复消费 → 所以唯一ID去重表/状态机/乐观锁
顺序消费	因为同一业务的消息乱序会出错 → 所以sharding key路由到同一队列
死信队列	因为重试3次还是失败 → 所以进入死信队列等待人工/自动补偿
延迟队列	因为业务需要"30分钟后执行" → 所以TTL+死信/RocketMQ延迟级别
MQ选型	因为需求不同 → Kafka(吞吐)/RabbitMQ(可靠)/RocketMQ(全功能)
消息积压	因为消费速度<生产速度 → 所以扩容消费者+跳过非核心+监控告警

📌 因果链总结

同步
阻塞 → 引入MQ → 消息
丢失 → 确认+
持久化 → 消息
重复 → 幂等
设计 → 顺序
要求 → 顺序
消费 → 消费
失败 → 死信
队列 → 积压 → 扩容
应急

全栈资料库 · 因果链板块 · 消息队列因果链