diff --git a/notes/消息队列.md b/notes/消息队列.md index a8cbf167..0dc4bdbf 100644 --- a/notes/消息队列.md +++ b/notes/消息队列.md @@ -23,6 +23,10 @@ * [为什么吞吐量高](#为什么吞吐量高) * [副本与可靠性](#副本与可靠性) * [附录：实测](#附录实测) + * [A1 批量对吞吐量的影响](#a1-批量对吞吐量的影响) + * [A2 重复消费](#a2-重复消费) + * [A3 消费积压](#a3-消费积压) + * [A4 Redis Stream 的确认](#a4-redis-stream-的确认) * [参考资料](#参考资料) @@ -78,7 +82,17 @@ Kafka、RocketMQ 等通过消费者组把两种模型结合起来：一条消息 ## 三、可靠性 -一条消息要经过发送、存储、消费三个环节，每个环节都可能丢失消息。 +一条消息要经过发送、存储、消费三个环节，每个环节都可能丢失消息。三种常用的消息队列在这三个环节上的做法： + +| 环节 | Kafka | RabbitMQ | RocketMQ（4.x） | +| --- | --- | --- | --- | +| 发送 | acks=all，失败自动重试 | 开启发布确认（Publisher Confirm），收到 ack 才算成功，并设置 mandatory | 同步发送，返回 SEND_OK 才算成功 | +| 存储 | 多副本，配合 min.insync.replicas，见 [副本与可靠性](#副本与可靠性) | 持久化的队列加持久化的消息，或者使用多副本的仲裁队列（Quorum Queue） | 同步刷盘（SYNC_FLUSH）加同步复制（SYNC_MASTER），默认都是异步的 | +| 消费 | 处理完再提交偏移量 | 关闭自动确认，处理完再发送 ack | 处理完再返回 CONSUME_SUCCESS，失败返回 RECONSUME_LATER 稍后重试 | + +需要注意的是，RabbitMQ 对无法路由到任何队列的消息同样返回 ack，只开启确认而不设置 mandatory 时，发错路由键的消息会被直接丢弃；设置了 mandatory，这类消息会先退回给生产者。 + +即使这些配置都打开，消息仍可能因为程序缺陷、误操作等原因丢失或者处理失败，关键业务还要定期核对上下游的数据（对账），发现缺失时补发。 ### 发送端的可靠性 @@ -101,17 +115,53 @@ Kafka、RocketMQ 等通过消费者组把两种模型结合起来：一条消息 - 保证接收端处理消息的业务逻辑具有幂等性：只要具有幂等性，那么消费多少次消息，最后处理的结果都是一样的； - 保证消息具有唯一编号，并使用一张日志表来记录已经消费的消息编号。 +例如积分服务消费“订单 1024 支付成功”的消息，在同一个本地事务中先写去重表、再加积分。重复的消息插入去重表时违反主键约束，回滚后直接确认这条消息： + +```sql +BEGIN; +INSERT INTO consumed_msg (msg_key) VALUES ('order:1024:paid'); -- 重复时报错 1062 +UPDATE points SET balance = balance + 100 WHERE user_id = 1024; +COMMIT; +``` + +去重的键要用业务上的唯一标识，例如订单号加事件类型，而不是消息队列生成的消息 ID。RocketMQ 文档指出，同一条消息被生产者重发后可能有两个不同的消息 ID。 + +去重记录也可以放在 Redis 中，但如果用 SETNX 标记之后、处理完成之前消费者崩溃，重新投递的消息会被当作已经处理而跳过，这条消息就丢了。因此 Redis 中要区分“处理中”和“已完成”两种状态，“处理中”设置过期时间，过期后允许重新处理。通用的幂等方法见 [分布式中的幂等](分布式.md#九幂等)。 + ## 四、顺序性 消息队列通常只保证单个队列（Kafka 中的分区）内的消息有序，多个队列之间没有顺序保证。并且在点对点模式下，一个消费者处理失败的消息会被重新投递给另一个消费者，顺序也会被打乱。 要保证同一业务的消息有序，需要让发送端把这些消息按业务 Key（例如订单号）路由到同一个队列，并且接收端对同一个队列只用一个线程顺序消费。 +RabbitMQ 中一个队列内的消息是先进先出的，但一个队列有多个消费者时，消息被拒绝后重新入队会打乱顺序。需要顺序时按业务 Key 把消息路由到多个队列，每个队列只有一个消费者；或者开启单活跃消费者（Single Active Consumer），同一时刻只有一个消费者接收消息，它断开后才由另一个消费者接替。Kafka 中这两件事由分区和消费者组完成：同一个 key 进入同一个分区，一个分区只分配给组内的一个消费者。 + ## 五、消息堆积 -消费速度长期低于生产速度时，消息会在队列中堆积。 +消费速度长期低于生产速度时，消息会在队列中堆积。堆积的程度用积压量（Lag）衡量，Kafka 中每个分区的积压量等于日志末尾的偏移量减去消费者组已提交的偏移量。例如积分服务只消费了 3000 条就停止，用 kafka-consumer-groups.sh 查看（Kafka 4.4 实测，省略了后三列，见[附录 A3](#a3-消费积压)）： -解决方法是增加消费者数量来提高消费能力，但是消费者数量超过队列（分区）数量之后就不再有效，需要同时增加队列数量。 +```text +GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG +points points-events 0 0 9237 9237 +points points-events 1 0 9577 9577 +points points-events 2 3000 11186 8186 +``` + +排查时先看积压是生产突增造成的，还是消费变慢造成的： + +- 生产突增：例如大促、批量任务，生产速率的监控会明显上升，峰值过后通常能自然消化； +- 消费变慢：最常见的原因是消费逻辑依赖的数据库或外部接口变慢，其次是某些消息反复处理失败、不断重试，或者处理超时导致消费者被踢出、频繁重平衡； +- 只有个别分区积压：通常是 key 分布不均，大量消息集中到了同一个分区。 + +处理方法： + +- 先解决消费变慢的原因，例如修复下游、把反复失败的消息转入死信队列； +- 增加消费者数量，但消费者数量超过分区（队列）数量之后不再有效。上面的主题有 3 个分区，组内启动 4 个消费者后，其中一个分到 0 个分区； +- 提高单个消费者的处理能力：批量写库，或者在消费者内部用线程池并行处理，同一个 key 的消息交给同一个线程以保持顺序； +- 积压量很大、分区数又不够时，临时新建一个分区更多的主题，原来的消费者只把消息原样转发过去，再用更多的消费者消费新主题； +- 不重要的消息可以降级，例如先跳过，之后再补偿。 + +需要注意的是，Kafka 按保留时间删除旧数据，不管消息有没有被消费，积压超过 retention.ms（默认 7 天）的消息会被删除；RocketMQ 默认保留 72 小时。积压的时间接近保留时间时，要先把消息转存出来。 ## 六、延迟消息与死信队列 @@ -152,6 +202,26 @@ JDK 自带的 DelayQueue 也能实现延迟，但数据只在单个进程的内 ## 七、Kafka +Kafka 把消息保存为按分区追加的日志，消息被消费后不会删除，这是它与 RabbitMQ 最大的区别。三种常用的消息队列对比如下： + +| | Kafka | RabbitMQ | RocketMQ | +| --- | --- | --- | --- | +| 存储 | 分区日志，按保留时间删除，与是否被消费无关 | 队列，消息被确认后删除 | 所有主题的消息顺序写入同一个 CommitLog，每个队列另有消费索引，按保留时间删除 | +| 消费 | 消费者拉取，提交偏移量，可以把偏移量调回去重新消费 | Broker 主动推送给消费者，用 prefetch 限制未确认的消息数 | 消费者按队列拉取（Push 模式用长轮询实现），可以按时间重置消费位置 | +| 路由 | 生产者按 key 选择分区 | 交换机按路由键把消息投递到一个或多个队列，规则灵活 | 主题加 Tag 过滤 | +| 业务功能 | 本身不支持延迟消息和死信 | 死信交换机、TTL、优先级队列，延迟消息需要插件 | 延迟消息、事务消息、消费重试和死信队列 | +| 适用场景 | 日志、埋点、数据同步、流处理等吞吐量大的场景 | 消息量不大、路由规则复杂的业务消息 | 订单、支付等需要事务消息、延迟消息和重试的业务消息 | + +RabbitMQ 也提供了只追加的 Stream 类型，消息确认后不删除，可以重复读取。 + +Redis 也常被用作轻量的消息队列，但可靠性不如专门的消息队列： + +- Pub/Sub 不保存消息，发布时没有订阅者或者订阅者断线，消息就丢了，实测没有订阅者时 PUBLISH 返回 0； +- List 用 LPUSH 和 BRPOP 实现队列，消息取出后就从 List 中删除，消费者处理时崩溃就丢了，也不支持多个消费者组； +- Stream（Redis 5.0 起）支持消费者组和确认：XREADGROUP 读出的消息进入待确认列表，XACK 之后才移除，消费者崩溃时其它消费者可以用 XAUTOCLAIM 接管它没有确认的消息（见[附录 A4](#a4-redis-stream-的确认)）。 + +即使使用 Stream，数据也保存在内存中，持久化和主从复制都是异步的，宕机时可能丢失最近写入的消息，能保存的消息量也受内存限制。消息量不大、能接受少量丢失时可以用 Redis，否则用专门的消息队列。 + ### 分区与消费者组 Kafka 中的消息按主题（Topic）分类，每个主题分成多个分区（Partition）。分区是一个只能在末尾追加的日志，每条消息在分区内有一个递增的偏移量（Offset），消费者记录自己读到了哪个偏移量。 @@ -193,12 +263,16 @@ Kafka 中的消息按主题（Topic）分类，每个主题分成多个分区（ 顺序写和页缓存：消息只追加到日志的末尾，是顺序写，比随机写快得多。写入时只写到操作系统的页缓存，由操作系统决定何时刷盘，消费者读取刚写入不久的数据时也通常直接命中页缓存。可靠性不依靠每条消息都刷盘，而是依靠多个副本。 -批量：生产者把发往同一个分区的消息攒成批次再发送，批次在网络传输、写入日志和副本复制时都作为一个整体处理，压缩也以批次为单位。例如在 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 实测，见[附录：实测](#附录实测)）。 +批量：生产者把发往同一个分区的消息攒成批次再发送，批次在网络传输、写入日志和副本复制时都作为一个整体处理，压缩也以批次为单位。例如在 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-批量对吞吐量的影响)）。 -零拷贝：broker 把日志发给消费者时使用 sendfile 系统调用，数据不需要先读到用户空间再写回内核（启用 SSL 时数据要在用户空间加密，不能使用 sendfile）： +零拷贝：零拷贝（Zero-copy）指的是传输数据时，CPU 不在内核空间和用户空间之间来回拷贝数据。用 read() 和 write() 把文件发到网络时，数据从页缓存拷贝到应用的缓冲区，再拷贝到 Socket 缓冲区，共 4 次拷贝、2 次系统调用。broker 把日志发给消费者时使用 sendfile 系统调用，数据只在内核中传递（启用 SSL 时数据要在用户空间加密，不能使用 sendfile）：

+另一种减少拷贝的方法是内存映射（mmap）：把文件映射到进程的地址空间，进程读写这段内存就是直接读写页缓存，不需要 read() 和 write() 的拷贝。第一次访问某一页时触发缺页异常，由内核把这一页从磁盘读入页缓存，再建立映射。前文 Kafka 的索引文件、RocketMQ 的 CommitLog 都是用 mmap 读写的。 + +两者的适用场景不同：sendfile 把文件原样发给 Socket，程序接触不到数据；mmap 适合程序本身要频繁读写文件内容的场景。Linux 还提供了 splice，在两个文件描述符之间移动数据而不经过用户空间，要求其中一个是管道。 + 分区并行：一个主题的多个分区分布在不同的 broker 上，读写由多台机器共同承担。 ### 副本与可靠性 @@ -230,13 +304,15 @@ acks=all 要配合 min.insync.replicas 使用，否则 ISR 只剩 Leader 一个 生产者没有收到确认就会重试，这可能让同一条消息写入两次。Kafka 3.0 起默认开启幂等生产者：broker 给每个生产者分配一个 ID，生产者给发往每个分区的消息编上序号，broker 发现序号重复就丢弃。前面查看 .log 文件时，每个批次都带有 producerId: 1001 和 baseSequence 字段。幂等只保证单个分区内不重复；要让写入多个分区的消息同时成功或失败，需要使用事务，消费者设置 isolation.level=read_committed 后只读取已提交事务中的消息。 -消费者处理完消息后提交偏移量，下次从提交的位置继续读取。默认每 5 秒自动提交一次，已经处理但还没提交时崩溃，重启后就会重复消费。例如消费者读到 3 条扣款消息后被 kill -9，这时组的偏移量仍然为空；它在 session.timeout.ms（默认 45 秒）之后被移出消费者组，重新启动的消费者又收到了这 3 条消息（见[附录：实测](#附录实测)）。因此接收端必须做到幂等，方法见 [接收端的可靠性](#接收端的可靠性)。反过来，如果先提交偏移量再处理，处理失败的消息就丢失了。 +消费者处理完消息后提交偏移量，下次从提交的位置继续读取。默认每 5 秒自动提交一次，已经处理但还没提交时崩溃，重启后就会重复消费。例如消费者读到 3 条扣款消息后被 kill -9，这时组的偏移量仍然为空；它在 session.timeout.ms（默认 45 秒）之后被移出消费者组，重新启动的消费者又收到了这 3 条消息（见[附录 A2](#a2-重复消费)）。因此接收端必须做到幂等，方法见 [接收端的可靠性](#接收端的可靠性)。反过来，如果先提交偏移量再处理，处理失败的消息就丢失了。 ## 附录：实测 环境为 Apple M1 Pro、macOS，Kafka 4.4.0 的 3 个节点的 KRaft 集群运行在同一台机器上（端口 9091～9093，命令中的 $B 为这三个地址）。 -批量对吞吐量的影响：主题有 3 个分区、3 个副本，每种配置运行 3 次： +### A1 批量对吞吐量的影响 + +主题有 3 个分区、3 个副本，每种配置运行 3 次： ```text $ kafka-topics.sh --bootstrap-server $B --create --topic perf --partitions 3 --replication-factor 3 @@ -252,7 +328,9 @@ $ ...（同上） acks=all batch.size=0 linger.ms=0 20000 records sent, 6795.786612 records/sec (0.65 MB/sec), 1410.16 ms avg latency, 2648.00 ms max latency ``` -重复消费：消费者组 billing 使用默认的自动提交，读到 3 条消息后立即 kill -9，45 秒后再启动一个同组的消费者： +### A2 重复消费 + +消费者组 billing 使用默认的自动提交，读到 3 条消息后立即 kill -9，45 秒后再启动一个同组的消费者： ```text 第一个消费者： @@ -267,12 +345,79 @@ $ ...（同上） acks=all batch.size=0 linger.ms=0 Processed a total of 3 messages ``` +### A3 消费积压 + +主题 points-events 有 3 个分区、3 个副本。用 kafka-console-producer.sh 写入 3 万条带 key 的消息（key 为 997 个用户 ID），消费者组 points 用 kafka-console-consumer.sh --max-messages 3000 消费 3000 条后退出，再查看积压；之后同时启动 4 个同组的消费者，查看每个消费者分到的分区数： + +```text +$ kafka-consumer-groups.sh --bootstrap-server $B --describe --group points +Consumer group 'points' has no active members. + +GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG +points points-events 0 0 9237 9237 +points points-events 1 0 9577 9577 +points points-events 2 3000 11186 8186 + +$ kafka-consumer-groups.sh --bootstrap-server $B --describe --group points --members +GROUP CONSUMER-ID #PARTITIONS +points console-consumer-473682e9-43d9-4ade-98fb-045cbad44ef5 1 +points console-consumer-45a48e35-9625-4cb2-8495-73c2f2da0830 1 +points console-consumer-dcea488d-0498-4116-8265-3aed53a8667d 0 +points console-consumer-610451bd-692d-45f0-8b8b-d0af64cda4b1 1 +``` + +### A4 Redis Stream 的确认 + +Redis 7.4.1，依次执行以下命令（redis-cli 非交互模式的原始输出，每个元素一行，空行表示空结果）： + +```text +$ redis-cli xadd s3:orders '*' orderId 1024 +1793162053370-0 +$ redis-cli xgroup create s3:orders points 0 +OK +$ redis-cli xreadgroup group points c1 count 10 streams s3:orders '>' +s3:orders +1793162053370-0 +orderId +1024 +$ redis-cli xpending s3:orders points # c1 读出但还没有确认 +1 +1793162053370-0 +1793162053370-0 +c1 +1 +$ redis-cli xreadgroup group points c2 count 10 streams s3:orders '>' + # 新消息已经被 c1 领走，c2 读不到 +$ redis-cli xautoclaim s3:orders points c2 0 0 # c2 接管 c1 没有确认的消息 +0-0 +1793162053370-0 +orderId +1024 + +$ redis-cli xack s3:orders points 1793162053370-0 +1 +$ redis-cli xpending s3:orders points +0 + + + +$ redis-cli publish ch:orders 1024 # 没有订阅者 +0 +``` + ## 参考资料 - [Designing Data-Intensive Applications 中文版](https://github.com/vonng/ddia) - [消息队列中点对点与发布订阅区别](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) +- [Consumer Acknowledgements and Publisher Confirms | RabbitMQ](https://www.rabbitmq.com/docs/confirms) +- [Queues: Message Ordering | RabbitMQ](https://www.rabbitmq.com/docs/queues) +- [最佳实践 | RocketMQ 4.x](https://rocketmq.apache.org/zh/docs/4.x/bestPractice/01bestpractice/) +- [RocketMQ 设计](https://github.com/apache/rocketmq/blob/develop/docs/cn/design.md) +- [Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/) +- [Redis Pub/Sub](https://redis.io/docs/latest/develop/pubsub/) +- [sendfile(2)](https://man7.org/linux/man-pages/man2/sendfile.2.html)、[splice(2)](https://man7.org/linux/man-pages/man2/splice.2.html)、[mmap(2)](https://man7.org/linux/man-pages/man2/mmap.2.html) - [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/)