Update 消息队列.md

补充延迟消息、死信队列和 Kafka。
2026-10-29 15:13:44 +00:00 · 2026-10-26 22:04:09 +08:00
parent d57319b29e
commit e06cd86190
3 changed files with 133 additions and 1 deletions
@@ -14,6 +14,14 @@
        * [接收端的可靠性](#接收端的可靠性)
    * [四、顺序性](#四顺序性)
    * [五、消息堆积](#五消息堆积)
+    * [六、延迟消息与死信队列](#六延迟消息与死信队列)
+        * [延迟消息](#延迟消息)
+        * [死信队列](#死信队列)
+    * [七、Kafka](#七kafka)
+        * [分区与消费者组](#分区与消费者组)
+        * [存储](#存储)
+        * [为什么吞吐量高](#为什么吞吐量高)
+        * [副本与可靠性](#副本与可靠性)
    * [参考资料](#参考资料)
 <!-- GFM-TOC -->

@@ -104,8 +112,132 @@ Kafka、RocketMQ 等通过消费者组把两种模型结合起来：一条消息

 解决方法是增加消费者数量来提高消费能力，但是消费者数量超过队列（分区）数量之后就不再有效，需要同时增加队列数量。

+## 六、延迟消息与死信队列
+
+### 延迟消息
+
+延迟消息指的是发送后不立即投递，等到指定时间才能被消费的消息。典型的场景是订单 30 分钟未支付自动取消：下单时发送一条 30 分钟后投递的消息，到时消费者检查订单，仍未支付就取消。相比定时任务扫描全表，延迟消息只处理到期的订单。
+
+各消息队列的支持情况：
+
+- RocketMQ 4.x 只支持 18 个固定的延迟级别：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，消息先存入内部的定时主题，到期后再投递到目标主题；
+- RocketMQ 5.x 支持指定任意的投递时间，默认最长 24 小时，精度为秒级；
+- RabbitMQ 没有原生的延迟消息，常用的做法是给消息设置过期时间（TTL），过期后转入死信交换机，由死信队列的消费者处理，或者安装延迟消息插件；
+- Kafka 不支持延迟消息，需要自己实现，例如按延迟时长分成几个主题，消费者读到未到期的消息就暂停等待。
+
+不使用消息队列时，也可以用 Redis 的有序集合实现：分值为到期时间，后台线程定期取出分值不大于当前时间的成员。多个实例同时轮询时，用 ZREM 的返回值判断由谁处理，返回 1 的实例才处理：
+
+```text
+> zadd delay:orders 1792997618 order:1024 1792999423 order:1025
+(integer) 2
+> zrangebyscore delay:orders 0 1792997623 limit 0 10
+1) "order:1024"
+> zrem delay:orders order:1024
+(integer) 1
+> zrem delay:orders order:1024
+(integer) 0
+```
+
+JDK 自带的 DelayQueue 也能实现延迟，但数据只在单个进程的内存中，服务重启就会丢失，多个实例之间也无法共享。
+
+### 死信队列
+
+消息多次消费失败后，不再重试，而是转入死信队列，避免一条有问题的消息反复重试、阻塞其它消息，之后由人工或专门的程序排查处理。
+
+- RocketMQ 并发消费时默认重试 16 次，仍然失败就进入名为 %DLQ% 加消费者组名的死信主题；
+- RabbitMQ 中消息被消费者拒绝且不重新入队、消息的 TTL 过期、队列长度超过上限时，消息会被转发到死信交换机。
+
+事务消息用于保证本地事务和发送消息同时成功或失败，见 [分布式事务中的事务消息](分布式.md#本地消息表)。
+
+## 七、Kafka
+
+### 分区与消费者组
+
+Kafka 中的消息按主题（Topic）分类，每个主题分成多个分区（Partition）。分区是一个只能在末尾追加的日志，每条消息在分区内有一个递增的偏移量（Offset），消费者记录自己读到了哪个偏移量。
+
+生产者发送消息时可以带一个 key，默认按 key 的哈希值对分区数取模选择分区，相同 key 的消息一定进入同一个分区，所以同一个订单的消息是有序的，不同分区之间则没有顺序。没有 key 时，生产者先把一批消息发往同一个分区，再换下一个分区，以便凑成批次。
+
+<div align="center"> <img src="pics/kafka-topic.png" width="760px"/> </div><br>
+
+消费者以消费者组为单位订阅主题：组内每个分区只分配给一个消费者，一个消费者可以负责多个分区。所以分区数决定了一个组最多有多少个消费者同时工作，多出来的消费者只能空闲。不同的组各自独立地消费全部消息，例如计费服务和物流服务各用一个组，都能收到每一条订单消息。
+
+消费者加入、退出或者分区数变化时，组内会重新分配分区，称为重平衡（Rebalance）。重平衡期间被重新分配的分区暂停消费，还可能造成重复消费。一个常见的原因是处理一批消息的时间超过了 max.poll.interval.ms（默认 5 分钟），消费者被认为已经失效，它的分区被分给其他消费者。
+
+### 存储
+
+每个分区在磁盘上是一个目录，日志按大小切分成多个段（Segment），文件名是段中第一条消息的偏移量。本机把段的大小设为 1 MB（默认 1 GB），向一个只有一个分区的主题写入 3 万条 100 字节的消息后，目录中的文件和大小如下（省略了 .timeindex 等文件）：
+
+```text
+00000000000000000000.log     1041728
+00000000000000000000.index       504
+00000000000000009472.log     1041728
+00000000000000009472.index       504
+00000000000000018944.log     1041728
+00000000000000018944.index       504
+00000000000000028416.log      174207
+00000000000000028416.index  10485760
+```
+
+.index 是稀疏索引，每写入一段数据（index.interval.bytes，默认 4 KB）记录一条“偏移量 → 在 .log 文件中的位置”。查找偏移量 20000 的消息：
+
+1. 按文件名二分查找，20000 落在 00000000000000018944 这个段中；
+2. 在这个段的索引中二分查找，不大于 20000 的最近一条索引是 19979 → 97662；
+3. 从 .log 文件的位置 97662 开始顺序读，下一个批次（偏移量 19980～20127）中就有 20000。
+
+正在写入的段的索引文件预先分配了 10 MB，并用 mmap 映射到内存，段写满后才截断为实际大小。用 kafka-dump-log.sh 查看 .log 文件，可以看到消息是按批次存放的，这里每个批次有 148 条消息、16277 字节，接近生产者批次大小的默认值 16 KB。
+
+旧数据按保留时间（默认 7 天）或总大小清理，每次删除一整个段，不需要修改文件内容；也可以配置为日志压缩，每个 key 只保留最新的一条消息。
+
+### 为什么吞吐量高
+
+顺序写和页缓存：消息只追加到日志的末尾，是顺序写，比随机写快得多。写入时只写到操作系统的页缓存，由操作系统决定何时刷盘，消费者读取刚写入不久的数据时也通常直接命中页缓存。可靠性不依靠每条消息都刷盘，而是依靠多个副本。
+
+批量：生产者把发往同一个分区的消息攒成批次再发送，批次在网络传输、写入日志和副本复制时都作为一个整体处理，压缩也以批次为单位。在本机 3 个节点的集群上用 kafka-producer-perf-test.sh 发送 2 万条 100 字节的消息（acks=all），各运行 3 次：默认配置每秒约 5.7 万～6.8 万条，平均延迟 21～54 ms；设置 batch.size=0 关闭批量后，每秒只有约 6500～6800 条，发送的请求大量排队，平均延迟约 1.4 秒。
+
+零拷贝：broker 把日志发给消费者时使用 sendfile 系统调用，数据不需要先读到用户空间再写回内核：
+
+<div align="center"> <img src="pics/zero-copy.png" width="760px"/> </div><br>
+
+分区并行：一个主题的多个分区分布在不同的 broker 上，读写由多台机器共同承担。
+
+### 副本与可靠性
+
+每个分区有多个副本，分布在不同的 broker 上，其中一个是 Leader，负责处理读写，其余的 Follower 不断从 Leader 拉取数据。与 Leader 保持同步的副本集合称为 ISR（In-Sync Replicas），Follower 超过 replica.lag.time.max.ms（默认 30 秒）没有追上 Leader 就会被移出 ISR。Kafka 4.0 起完全移除了 ZooKeeper，Leader 选举等元数据由内置的 KRaft 控制器管理，它基于 Raft 协议实现。
+
+本机创建 3 个分区、3 个副本的主题，然后用 kill -9 停掉 broker 3：
+
+```text
+停止前
+Partition: 0    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1
+Partition: 1    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2
+Partition: 2    Leader: 1    Replicas: 1,2,3    Isr: 1,2,3
+停止后
+Partition: 0    Leader: 2    Replicas: 2,3,1    Isr: 2,1
+Partition: 1    Leader: 1    Replicas: 3,1,2    Isr: 1,2
+Partition: 2    Leader: 1    Replicas: 1,2,3    Isr: 1,2
+```
+
+分区 1 的 Leader 从 broker 3 切换到了 ISR 中的 broker 1，其它分区只是 ISR 中少了 3。broker 3 重启并追上数据后，又回到了 ISR 中。
+
+生产者的 acks 参数决定写入多少个副本才算成功：
+
+- acks=0：不等待确认，可能丢消息；
+- acks=1：Leader 写入就返回，Leader 宕机而 Follower 还没同步时会丢消息；
+- acks=all：ISR 中的副本都写入才返回，Kafka 3.0 起是默认值。
+
+acks=all 要配合 min.insync.replicas 使用，否则 ISR 只剩 Leader 一个时，acks=all 就退化成了 acks=1。min.insync.replicas 设为 2 时，ISR 少于 2 个副本，broker 就拒绝 acks=all 的写入。上面停掉 broker 3 之后 ISR 还有 2 个副本，写入正常；把 min.insync.replicas 改为 3 再写入，生产者重试几次后报错 NotEnoughReplicasException，改用 acks=1 则仍然能写入。常用的配置是 3 个副本、min.insync.replicas=2、acks=all，一个副本故障时既不丢消息，也不停止写入。
+
+生产者没有收到确认就会重试，这可能让同一条消息写入两次。Kafka 3.0 起默认开启幂等生产者：broker 给每个生产者分配一个 ID，生产者给发往每个分区的消息编上序号，broker 发现序号重复就丢弃。前面查看 .log 文件时，每个批次都带有 producerId: 1001 和 baseSequence 字段。幂等只保证单个分区内不重复；要让写入多个分区的消息同时成功或失败，需要使用事务，消费者设置 isolation.level=read_committed 后只读取已提交事务中的消息。
+
+消费者处理完消息后提交偏移量，下次从提交的位置继续读取。默认每 5 秒自动提交一次，已经处理但还没提交时崩溃，重启后就会重复消费。本机实验中，消费者读到 3 条扣款消息后被 kill -9，这时组的偏移量仍然为空；它在 session.timeout.ms（默认 45 秒）之后被移出消费者组，重新启动的消费者又收到了这 3 条消息。因此接收端必须做到幂等，方法见 [接收端的可靠性](#接收端的可靠性)。反过来，如果先提交偏移量再处理，处理失败的消息就丢失了。
+
 ## 参考资料

 - [Observer vs Pub-Sub](http://developers-club.com/posts/270339/)
 - [Designing Data-Intensive Applications 中文版](https://github.com/vonng/ddia)
- [消息队列中点对点与发布订阅区别](https://blog.csdn.net/lizhitao/article/details/47723105)
+- [消息队列中点对点与发布订阅区别](https://web.archive.org/web/20250522024511/https://blog.csdn.net/lizhitao/article/details/47723105)
+- [定时/延时消息 | RocketMQ](https://rocketmq.apache.org/zh/docs/featureBehavior/02delaymessage/)
+- [Dead Letter Exchanges | RabbitMQ](https://www.rabbitmq.com/docs/dlx)
+- [Apache Kafka Documentation: Design](https://kafka.apache.org/43/design/design/)
+- [Apache Kafka Documentation: Configuration](https://kafka.apache.org/43/configuration/)
+- [Apache Kafka Documentation: Upgrading](https://kafka.apache.org/43/getting-started/upgrade/)