Update 消息队列.md

补充消费确认与重新投递、顺序消费的失败处理、延迟消息与时间轮、topic 交换机的匹配规则、Redis Stream 的可靠性、mmap 的拷贝次数等说明。

notes/消息队列.md

@@ -139,7 +139,11 @@ Kafka 默认不在每次写入后刷盘（log.flush.interval.messages 默认为
 
 ### 接收端的可靠性
 
-接收端在处理完成之后才向消息队列确认，处理过程中崩溃的话消息会被重新投递，不会丢失，代价是同一条消息可能被处理多次。生产者重发、消费者处理完但没来得及确认、重平衡后分区换了消费者，都会造成重复，Kafka 的幂等生产者只能去掉生产者内部重试造成的重复。所以处理消息的逻辑要具有幂等性，同一条消息处理多次和处理一次的结果相同，常用的做法是用一张去重表记录已经处理过的消息的业务键。
+接收端在处理完成之后才向消息队列确认，处理过程中崩溃的话消息会被重新投递，不会丢失，代价是同一条消息可能被处理多次。生产者重发、消费者处理完但没来得及确认、重平衡后分区换了消费者，都会造成重复，Kafka 的幂等生产者只能去掉生产者内部重试造成的重复。
+
+RocketMQ 5.x 的 SimpleConsumer 取消息时设定不可见时间，到期仍未确认的消息会重新投递，可能由组内其它消费者收到，原消费者之后再确认会失败；预计处理不完时，要在到期前调用 changeInvisibleDuration 延长。
+
+所以处理消息的逻辑要具有幂等性，同一条消息处理多次和处理一次的结果相同，常用的做法是用一张去重表记录已经处理过的消息的业务键。
 
 例如积分服务消费“订单 1024 支付成功”的消息，在同一个本地事务中先写去重表、再加积分。重复的消息插入去重表时违反主键约束，回滚后直接确认这条消息：
 
@@ -172,7 +176,7 @@ RabbitMQ 中一个队列内的消息是先进先出的，但一个队列有多
 
 需要注意的是，分区数改变后同一个 key 可能落到另一个分区。例如主题从 3 个分区扩为 4 个后，订单 1024 的“已支付”在分区 2，“已发货”进入了分区 1，可能先被处理（Kafka 4.4 实测，见[附录 A5](#a5-增加分区后-key-所在的分区)）。所以要求顺序的主题要预先规划分区数，扩容时先暂停写入，等旧分区消费完再恢复。
 
-顺序消费时一条消息处理失败，后面的消息都要等它。RocketMQ 4.x 的顺序消费默认一直重试，要设置最大重试次数（maxReconsumeTimes），超过后转入[死信队列](#死信队列)。
+顺序消费时一条消息处理失败，后面的消息都要等它。RocketMQ 4.x 的顺序消费默认一直重试，要设置最大重试次数（maxReconsumeTimes），超过后转入[死信队列](#死信队列)。5.x 的新 SDK 按消息组（MessageGroup）保证顺序，同一消息组的消息存放在同一个队列中；一条消息超过最大重试次数后跳过，不再阻塞后面的消息（RocketMQ 5.x 文档）。
 
 ## 五、消息堆积 🔥
 
@@ -213,7 +217,7 @@ RabbitMQ 的队列默认没有长度上限，积压的消息占用内存和磁
 
 延迟消息指的是发送后不立即投递，等到指定时间才能被消费的消息。典型的场景是订单 30 分钟未支付自动取消：下单时发送一条 30 分钟后投递的消息，到时消费者检查订单，仍未支付就取消。相比[定时任务扫描全表](分布式.md#定时扫描与延迟消息)，延迟消息只处理到期的订单。
 
-延迟消息也可能丢失、晚到或重复，所以取消时以订单的当前状态为条件：UPDATE orders SET status = 'CANCELLED' WHERE id = 1024 AND status = 'UNPAID'，支付回调同样带状态条件，取消后才到的支付成功要退款；要求高的业务再用低频扫描兜底。取消前先向支付平台关单，用户就不能在取消后再付款，见[超时关单](系统设计基础.md#超时关单)。
+延迟消息也可能丢失、晚到或重复，所以取消时以订单的当前状态为条件：UPDATE orders SET status = 'CANCELLED' WHERE id = 1024 AND status = 'UNPAID'，支付回调同样带状态条件，取消后才到的支付成功要退款；要求高的业务再用低频扫描兜底，扫描时同时处理消息丢失和关单停在一半的订单。取消前先向支付平台关单，用户就不能在取消后再付款，见[超时关单](系统设计基础.md#超时关单)。
 
 各消息队列的支持情况：
 
@@ -237,13 +241,19 @@ RabbitMQ 的队列默认没有长度上限，积压的消息占用内存和磁
 
 需要注意的是，ZREM 之后、处理完成之前实例崩溃，任务就丢了；要求可靠时用 Lua 脚本把任务原子地移到“处理中”的有序集合，处理完再删除，超时未删除的放回。
 
+不要用键空间通知的 expired 事件做延迟触发。这个事件在键被删除时才产生，而不是在过期时间到达时；没有命令访问这个键、带过期时间的键又很多时，可能晚很久才删除。键空间通知基于 Pub/Sub，订阅端断线期间的事件也会丢失（Redis 键空间通知文档）。
+
 JDK 自带的 DelayQueue 也能实现延迟，但数据只在单个进程的内存中，服务重启就会丢失，多个实例之间也无法共享。
 
+DelayQueue 内部是按到期时间排序的堆，插入和取出都是 O(logN)。任务很多时常用时间轮（选读）：把一圈分成若干格，指针每隔固定时间走一格，任务按到期时间放进对应的格子，插入是 O(1)；超过一圈的任务记下还要转几圈，或者放进更粗粒度的上层时间轮。
+
+例如 Netty 的 HashedWheelTimer 默认每格 100 ms、一圈 512 格，计时器刚启动时提交一个 30 分钟的任务，要走 18000 格，18000 = 35 × 512 + 80，所以它放在第 80 格，还要再转 35 圈。HashedWheelTimer 只在进程的内存中保存任务，重启后会丢失；RocketMQ 5.x 的定时消息则把时间轮和消息一起持久化在 Broker 上。
+
 ### 死信队列
 
 消息多次消费失败后，不再重试，而是转入死信队列，避免一条有问题的消息反复重试、阻塞其它消息，之后由人工或专门的程序排查，修复后重新投递。重试间隔通常逐次拉长，因为失败多由下游故障引起，立即重试多半还会失败。
 
-- RocketMQ 并发消费时默认重试 16 次，间隔从 10s 逐步拉长到 2h，仍然失败就进入名为 %DLQ% 加消费者组名的死信主题，死信同样只保存 72 小时；
+- RocketMQ 4.x 客户端并发消费时默认重试 16 次，间隔从 10s 逐步拉长到 2h，仍然失败就进入名为 %DLQ% 加消费者组名的死信主题，死信同样只保存 72 小时；
 - RabbitMQ 中消息被消费者拒绝且不重新入队、消息的 TTL 过期、队列长度超过上限时，消息会被转发到死信交换机。仲裁队列中投递失败超过 delivery-limit 次（4.0 起默认 20 次）的消息也会转入，没有配置死信交换机就直接删除；
 - Kafka 的普通消费者组没有死信，由消费者把多次失败的消息写到重试主题、死信主题，例如 Spring Kafka 的 DeadLetterPublishingRecoverer。
 
@@ -263,6 +273,8 @@ Kafka 把消息保存为按分区追加的日志，消息被消费后不会删
 
 Kafka 选择拉模式，是因为推送的速率由 Broker 决定，需要额外的流控（例如 RabbitMQ 的 prefetch）才能匹配消费者的处理速度；拉取时消费者按自己的速度读取，一次可以拉走当前位置之后的一批消息。没有新消息时，拉取请求在 Broker 上等待数据到达（长轮询），避免空转。
 
+RabbitMQ 的 topic 交换机把路由键按点分成单词，绑定键中 * 匹配一个单词，# 匹配零个或多个单词。例如绑定键 order.# 能匹配 order、order.paid 和 order.paid.cn，order.* 只匹配 order.paid 这样两个单词的键。
+
 RabbitMQ 也提供了只追加的 Stream 类型，消息确认后不删除，可以重复读取。
 
 选型时先看业务需要什么：
@@ -279,7 +291,13 @@ Redis 也常被用作轻量的消息队列，但可靠性不如专门的消息
 - List 用 LPUSH 和 BRPOP 实现队列，消息取出后就从 List 中删除，消费者处理时崩溃就丢了，也不支持多个消费者组；
 - Stream（Redis 5.0 起）支持消费者组和确认：XREADGROUP 读出的消息进入待确认列表，XACK 之后才移除，消费者崩溃时其它消费者可以用 XCLAIM（6.2 起也可用 XAUTOCLAIM）接管它没有确认的消息（见[附录 A4](#a4-redis-stream-的确认)）。
 
-即使使用 Stream，数据也保存在内存中，默认配置下持久化和主从复制都是异步的，宕机时可能丢失最近写入的消息，能保存的消息量也受内存限制。消息量不大、能接受少量丢失时可以用 Redis，否则用专门的消息队列。
+即使使用 Stream，默认配置下持久化和主从复制都是异步的，宕机时可能丢失最近写入的消息。
+
+数据保存在内存中，能保存的消息量受内存限制。用 MAXLEN 限制长度时，还没有消费或确认的消息也会被裁掉，Redis 8.2 起可以加 ACKED 选项只裁所有消费者组都确认过的消息。
+
+同一条消息也可能被处理两次：消费者处理完、确认前崩溃，消息会被其它消费者接管；还没复制到从节点的 XACK 也可能在主从切换时丢失。所以消费端仍要幂等。
+
+消息量不大、能接受少量丢失时可以用 Redis，否则用专门的消息队列。
 
 ### 分区与消费者组
 
@@ -295,7 +313,9 @@ Kafka 4.2 起提供共享组（Share Group）：一个分区的消息可以分
 
 消费者加入、退出或者分区数变化时，组内会重新分配分区，称为重平衡（Rebalance）。默认的 classic 协议使用 RangeAssignor 分配，所有消费者先交出全部分区，整个组暂停；classic 下改用 CooperativeStickyAssignor，或者使用 Kafka 4.0 起的新协议（group.protocol=consumer，由 broker 计算分配），都只暂停需要移动的分区。例如 6 个分区的组加入第二个消费者时，第一个消费者在两种协议下分别交出 6 个和 3 个分区（Kafka 4.4 实测，见[附录 A7](#a7-重平衡时交出的分区)）。
 
-重平衡还可能造成重复消费。一个常见的原因是处理一批消息的时间超过了 max.poll.interval.ms（默认 5 分钟），消费者被认为已经失效，它的分区被分给其他消费者，这时它再提交偏移量会失败。可以调小 max.poll.records 或调大 max.poll.interval.ms；正常的重平衡可以在 ConsumerRebalanceListener 的 onPartitionsRevoked 中先同步提交已处理完的偏移量，再交出分区。
+重平衡还可能造成重复消费。一个常见的原因是处理一批消息的时间超过了 max.poll.interval.ms（默认 5 分钟），消费者被认为已经失效，它的分区被分给其他消费者，这时它再提交偏移量会失败。失去分区并不会中断正在执行的业务逻辑，旧消费者可能还在处理已经拉取的消息，与接管的消费者重叠，处理完再提交时才失败。可以调小 max.poll.records 或调大 max.poll.interval.ms；正常的重平衡可以在 ConsumerRebalanceListener 的 onPartitionsRevoked 中先同步提交已处理完的偏移量，再交出分区。
+
+发布新版本时，消费者收到 SIGTERM 后设置停止标志，并在另一个线程调用 wakeup()；消费线程处理完当前一批，处理好 WakeupException 后提交已完成的偏移量，再调用 close()。默认的动态成员调用 close() 时会主动离开消费者组，分区很快被接管；进程被 SIGKILL 强制终止时，通常要等会话超时后才会重新分配分区。
 
 ### 副本与可靠性
 
@@ -369,12 +389,14 @@ acks=all 要配合 min.insync.replicas（默认为 1）使用，否则 ISR 只
 
 批量：生产者把发往同一个分区的消息攒成批次再发送，批次在网络传输、写入日志和副本复制时都作为一个整体处理，压缩也以批次为单位。例如在 3 个节点的集群上发送 2 万条 100 字节的消息（acks=all），默认配置每秒约 5.7 万～6.8 万条，平均延迟 21～54 ms；设置 batch.size=0（linger.ms=0）关闭批量后，每秒只有约 6500～6800 条，发送的请求大量排队，平均延迟约 1.4 秒（Kafka 4.4 实测，见[附录 A1](#a1-批量对吞吐量的影响)）。
 
-零拷贝：零拷贝（Zero-copy）指的是传输数据时，CPU 不在内核空间和用户空间之间来回拷贝数据。用 read() 和 write() 把文件发到网络时，数据从页缓存拷贝到应用的缓冲区，再拷贝到 Socket 缓冲区，共 4 次拷贝、2 次系统调用。broker 把日志发给消费者时使用 sendfile 系统调用，数据只在内核中传递（启用 SSL 时数据要在用户空间加密，不能使用 sendfile）：
+零拷贝：零拷贝（Zero-copy）指的是传输数据时，CPU 不在内核空间和用户空间之间来回拷贝数据。用 read() 和 write() 把文件发到网络时，数据从页缓存拷贝到应用的缓冲区，再拷贝到 Socket 缓冲区，共 4 次拷贝、2 次系统调用。broker 把日志发给消费者时使用 sendfile 系统调用，数据只在内核中传递（启用 SSL，包括 SASL_SSL 时，数据要读到用户空间加密，不能使用 sendfile，但读的仍是页缓存；SASL_PLAINTEXT 只认证、不加密，仍使用 sendfile）：
 
 <div align="center"> <img src="pics/zero-copy.png" width="760px"/> </div><br>
 
 另一种减少拷贝的方法是内存映射（mmap）：把文件映射到进程的地址空间，进程读写这段内存就是直接读写页缓存，不需要 read() 和 write() 的拷贝。第一次访问某一页时触发缺页异常，由内核把这一页从磁盘读入页缓存，再建立映射。前文 Kafka 的索引文件、RocketMQ 的 CommitLog 都是用 mmap 读写的。
 
+用 mmap 映射文件后再调用 write() 发给 Socket 时，省掉的只是页缓存到应用缓冲区的一次拷贝，页缓存到 Socket 缓冲区仍要由 CPU 拷贝一次，共 3 次拷贝、2 次系统调用（mmap() 和 write()）。
+
 两者的适用场景不同：sendfile 把文件原样发给 Socket，程序接触不到数据；mmap 适合程序本身要频繁读写文件内容的场景。Linux 还提供了 splice，在两个文件描述符之间移动数据而不经过用户空间，要求其中一个是管道。
 
 分区并行：一个主题的多个分区分布在不同的 broker 上，读写由多台机器共同承担。