消息队列基础

1. 这是什么

消息队列用于在系统之间传递消息，实现异步、解耦和削峰。
它是中大型后端系统中非常常见的基础设施。

一句话理解：

• 消息队列不是为了“更酷”
• 它是为了让系统在时序、压力和依赖关系上更可控

2. 为什么重要

很多系统链路并不适合全部同步调用。
如果所有步骤都同步串在一起，就容易出现：

• 上游等待时间过长
• 下游抖动直接传染上游
• 高峰期整体雪崩

消息队列能帮助系统把：

• 高峰压力
• 跨服务流程
• 延迟任务

拆开处理。

3. 先建立直觉：消息队列的三大价值

3.1 异步

把不需要立刻完成的动作放到后面做。
例如：

• 下单成功后异步发短信

3.2 解耦

让生产者和消费者不必强绑定。
生产者只负责发消息，不必知道消费者细节。

3.3 削峰

当瞬时流量太高时，先把请求压进队列，再由消费者按能力逐步处理。
这能把尖峰流量摊平。

4. 核心内容

4.1 消息投递与消费

最基础的链路通常是：

1. 生产者发消息
2. 消息进入队列
3. 消费者拉取或接收消息
4. 消费者处理并确认

学习阶段更重要的是理解这个流程，而不是先死背某个 MQ 产品命令。

4.2 顺序消费

顺序消费指的是：

• 消息处理顺序必须和业务顺序一致

典型场景：

• 同一订单状态流转
• 同一账户余额变更

但顺序通常意味着：

• 并发自由度下降

所以只有在确实需要时才强调顺序。

4.3 重复消费

消息系统里，“至少一次投递”是很常见的现实语义。
这就意味着：

• 消息可能重复到达

所以消费者一侧通常必须有：

• 幂等设计

4.4 消息积压

消息积压指的是：

• 生产速度大于消费速度

它不是一定错误，但如果持续积压，就意味着：

• 消费能力不足
• 下游处理太慢
• 或消息设计不合理

4.5 消息丢失

消息丢失问题要从链路上看：

• 生产端有没有成功发送
• Broker 有没有持久化
• 消费端有没有正确确认

所以它不是一个单点概念，而是整条链的可靠性设计问题。

5. 学习重点

这一章最重要的是掌握这些判断：

• 消息队列不是为了“更快”，而是为了更合理的结构和压力管理
• 异步、解耦、削峰是三种不同收益
• 顺序、幂等、可靠性和积压处理是引入 MQ 后必须面对的新问题
• 用了 MQ 不代表一致性问题自动消失

6. 常见问题

6.1 把所有链路都强行改成异步

异步不是银弹。
如果业务本质上必须同步确认，那就不该硬改成异步。

6.2 没有幂等设计就直接消费消息

这是消息系统里非常危险的坑。

6.3 忽视消息积压后的系统连锁反应

积压不只是“队列里消息多一点”，它往往意味着：

• 业务延迟增加
• 补偿压力增加
• 下游资源继续恶化

7. 动手验证

这一节我用纯 Java 的阻塞队列做一个最小实验，帮助你直观看到“生产快、消费慢、队列缓冲”的核心效果。
它不是完整 MQ 产品，但非常适合理解本质。

7.1 准备一个可运行示例

新建文件 MessageQueueDemo.java，内容如下：

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class MessageQueueDemo {
    public static void main(String[] args) throws Exception {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
        AtomicInteger produced = new AtomicInteger(0);
        AtomicInteger consumed = new AtomicInteger(0);
        AtomicInteger maxBacklog = new AtomicInteger(0);

        Thread producer = new Thread(() -> {
            for (int i = 1; i <= 20; i++) {
                try {
                    queue.put("msg-" + i);
                    produced.incrementAndGet();
                    maxBacklog.updateAndGet(old -> Math.max(old, queue.size()));
                    System.out.println("produced=msg-" + i + ",queueSize=" + queue.size());
                    Thread.sleep(20);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        });

        Thread consumer = new Thread(() -> {
            for (int i = 1; i <= 20; i++) {
                try {
                    String msg = queue.take();
                    Thread.sleep(80);
                    consumed.incrementAndGet();
                    System.out.println("consumed=" + msg + ",queueSize=" + queue.size());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
        });

        producer.start();
        consumer.start();
        producer.join();
        consumer.join();

        System.out.println("producedTotal=" + produced.get());
        System.out.println("consumedTotal=" + consumed.get());
        System.out.println("maxBacklog=" + maxBacklog.get());
    }
}

7.2 编译并运行

javac MessageQueueDemo.java
java MessageQueueDemo

7.3 你应该观察到什么

输出会有很多行，但应包含这些关键信息：

producedTotal=20
consumedTotal=20
maxBacklog=...

中间你还会看到类似：

produced=msg-...
consumed=msg-...
queueSize=...

7.4 每一行在验证什么

• producedTotal=20：说明生产者总共发出了 20 条消息
• consumedTotal=20：说明消费者最终把消息处理完了
• maxBacklog=...：说明当生产快于消费时，消息会在队列中积压
• 中间的 queueSize 变化：直观展示了队列如何起到削峰缓冲作用

8. 练习建议

下面这些练习做完，这一章会更扎实：

• 做一个下单后异步通知的 demo
• 分析消息重复消费的处理方式
• 模拟消费端处理变慢时的积压现象
• 总结消息队列引入后的收益和代价

9. 自测问题

• 为什么消息队列适合削峰和解耦？
• 消息丢失和重复消费分别怎么处理？
• 什么场景不适合强行引入消息队列？
• 顺序消费为什么常常意味着更低并发？
• 为什么用了 MQ 之后幂等会变成高频要求？

10. 自测核对要点

如果你的回答能覆盖下面这些点，说明这一章基本掌握到位了：

• MQ 的核心价值在于异步、解耦、削峰
• 顺序、幂等、可靠性、积压是消息系统的关键配套问题
• 积压本质上反映了生产消费速率不匹配
• 引入消息队列是架构权衡，不是默认更优