Update 消息队列.md

补充削峰的方式、与 RPC 调用的区别、生产者的异步发送与刷盘、业务幂等与恰好一次、积压的监控、选型和并行消费时的偏移量提交。

notes/消息队列.md

@@ -8,6 +8,7 @@
         * [异步处理](#异步处理)
         * [流量削峰](#流量削峰)
         * [应用解耦](#应用解耦)
+        * [与 RPC 调用的区别](#与-rpc-调用的区别)
     * [三、可靠性](#三可靠性)
         * [发送端的可靠性](#发送端的可靠性)
         * [消息队列的可靠性](#消息队列的可靠性)
@@ -72,14 +73,26 @@ Kafka、RocketMQ 等通过消费者组把两种模型结合起来：一条消息
 
 可以将请求发送到消息队列中，服务器按照其处理能力从消息队列中订阅消息进行处理。
 
+“按处理能力”是由消费者控制的：Kafka 的消费者每次 poll 最多取回 max.poll.records（默认 500）条消息，处理完再取下一批；RabbitMQ 用 prefetch 限制推给每个消费者、还没有确认的消息数。峰值期间来不及处理的请求留在队列中，峰值过后再慢慢处理完，积压量和排空时间的估算见[削峰](系统设计基础.md#2-削峰)。
+
+削峰的代价是用户不能立即得到结果，前端要显示“排队中”，再查询处理结果。队列也不能无限积压，入口仍然需要限流：排队时间超过用户能接受的程度时，不如直接返回“已售罄”或“稍后再试”。
+
 ### 应用解耦
 
 如果模块之间不直接进行调用，模块之间耦合度就会很低，那么修改一个模块或者新增一个模块对其它模块的影响会很小，从而实现可扩展性。
 
 通过使用消息队列，一个模块只需要向消息队列中发送消息，其它模块可以选择性地从消息队列中订阅消息从而完成调用。
 
+例如订单支付成功后，积分、物流、短信和数据分析四个服务都要处理。订单服务逐个调用它们时，每新增一个下游都要修改订单服务，任何一个下游超时都会拖慢支付流程；改为发送一条“订单 1024 已支付”的消息后，各下游用自己的消费者组订阅，新增下游不需要修改订单服务，某个下游故障时消息留在队列中，恢复后继续处理。
+
 引入消息队列的代价是系统多了一个需要保证高可用的组件，并且需要处理下面的可靠性和顺序性问题。
 
+### 与 RPC 调用的区别
+
+RPC 是同步的请求-响应：调用方等待被调用方返回结果，被调用方必须在线，适合需要立即得到结果的调用；消息队列是异步的：发送方只确认消息已经存入队列，不知道也不等待谁来处理、何时处理，适合结果可以稍后产生、或者有多个接收方的场景。
+
+选择时可以问两个问题：调用方的下一步是否依赖这次调用的结果？被调用方暂时不可用时，调用方是否也应该失败？两个答案都是“是”时用 RPC，否则考虑消息队列。例如支付前必须知道库存是否扣减成功，用 RPC 调用库存服务；支付成功后加积分，积分服务晚几秒处理、甚至宕机几分钟都不影响支付，用消息队列。
+
 ## 三、可靠性
 
 一条消息要经过发送、存储、消费三个环节，每个环节都可能丢失消息。三种常用的消息队列在这三个环节上的做法：
@@ -102,10 +115,14 @@ Kafka、RocketMQ 等通过消费者组把两种模型结合起来：一条消息
 
 消息队列把消息可靠地保存之后才向发送端返回确认，发送端没有收到确认就进行重传。这里的“保存”不一定是刷盘，例如 Kafka 中是写入了足够多的副本，见下文的[副本与可靠性](#副本与可靠性)。
 
+需要注意的是，Kafka 生产者的 send() 是异步的，把消息放入本地的缓冲区就返回，发送和重试在后台线程中进行。既不检查返回的 Future、也不传入回调时，重试用完之后的失败没有任何人知道；进程退出前没有调用 close() 或 flush()，缓冲区中还没有发出的消息也会丢失。
+
 ### 消息队列的可靠性
 
 消息队列收到消息后需要持久化到磁盘，并复制到多个副本上，避免单个节点宕机导致消息丢失。
 
+Kafka 默认不在每次写入后刷盘（log.flush.interval.messages 默认为 Long 的最大值），持久性依靠多个副本。如果所有副本所在的机器同时断电，还在页缓存中的消息仍然会丢失，所以副本要分布在不同的机架或可用区。另一个相关的配置是 unclean.leader.election.enable，默认为 false：ISR 中的副本全部不可用时，宁可让分区暂时无法写入，也不让落后的副本成为 Leader；设为 true 可以更快恢复写入，但落后的副本没有同步到的消息就丢失了。
+
 ### 接收端的可靠性
 
 接收端能够从消息队列成功消费一次消息。
@@ -126,6 +143,10 @@ COMMIT;
 
 去重的键要用业务上的唯一标识，例如订单号加事件类型，而不是消息队列生成的消息 ID。RocketMQ 文档指出，同一条消息被生产者重发后可能有两个不同的消息 ID。
 
+有状态的业务也可以直接用状态做条件，不需要去重表。例如处理“订单 1024 已支付”的消息时执行 UPDATE orders SET status = 'PAID' WHERE id = 1024 AND status = 'UNPAID'，重复的消息影响行数为 0，直接确认即可。
+
+需要注意的是，消息队列提供的“恰好一次”（Exactly Once）不能代替业务的幂等。Kafka 的恰好一次指的是“从 Kafka 读取、处理、再写回 Kafka”这个过程中，用事务把输出的消息和消费的偏移量一起提交。消费者把结果写入数据库或者调用外部接口时，数据库的提交和偏移量的提交仍然是两件事，两者之间崩溃就会重复处理，Kafka 文档也说明写入其它系统时需要那个系统配合。
+
 去重记录也可以放在 Redis 中，但如果用 SETNX 标记之后、处理完成之前消费者崩溃，重新投递的消息会被当作已经处理而跳过，这条消息就丢了。因此 Redis 中要区分“处理中”和“已完成”两种状态，“处理中”设置过期时间，过期后允许重新处理。通用的幂等方法见 [分布式中的幂等](分布式.md#九幂等)。
 
 ## 四、顺序性
@@ -147,6 +168,8 @@ points   points-events   1          0               9577            9577
 points   points-events   2          3000            11186           8186
 ```
 
+积压要靠监控提前发现。常用的告警指标有两个：积压量持续增长，说明消费速率一直低于生产速率；消费延迟，即最早一条未消费的消息已经等待了多久，它与业务的关系更直接，例如订单超时取消的消息延迟超过 1 分钟就应该告警。
+
 排查时先看积压是生产突增造成的，还是消费变慢造成的：
 
 - 生产突增：例如大促、批量任务，生产速率的监控会明显上升，峰值过后通常能自然消化；
@@ -157,12 +180,14 @@ points   points-events   2          3000            11186           8186
 
 - 先解决消费变慢的原因，例如修复下游、把反复失败的消息转入死信队列；
 - 增加消费者数量，但消费者数量超过分区（队列）数量之后不再有效。上面的主题有 3 个分区，组内启动 4 个消费者后，其中一个分到 0 个分区；
-- 提高单个消费者的处理能力：批量写库，或者在消费者内部用线程池并行处理，同一个 key 的消息交给同一个线程以保持顺序；
+- 提高单个消费者的处理能力：批量写库，或者在消费者内部用线程池并行处理，同一个 key 的消息交给同一个线程以保持顺序，这时提交偏移量要注意的问题见[副本与可靠性](#副本与可靠性)的最后一段；
 - 积压量很大、分区数又不够时，临时新建一个分区更多的主题，原来的消费者只把消息原样转发过去，再用更多的消费者消费新主题；
 - 不重要的消息可以降级，例如先跳过，之后再补偿。
 
 需要注意的是，Kafka 按保留时间删除旧数据，不管消息有没有被消费，积压超过 retention.ms（默认 7 天）的消息会被删除；RocketMQ 默认保留 72 小时。积压的时间接近保留时间时，要先把消息转存出来。
 
+RabbitMQ 的队列默认没有长度上限，积压的消息占用内存和磁盘，达到内存或磁盘的水位线后，RabbitMQ 会阻塞所有发布消息的连接；设置了长度上限（x-max-length）的队列满了之后，默认丢弃队头最早的消息，配置了死信交换机时转入死信队列。
+
 ## 六、延迟消息与死信队列
 
 ### 延迟消息
@@ -214,6 +239,14 @@ Kafka 把消息保存为按分区追加的日志，消息被消费后不会删
 
 RabbitMQ 也提供了只追加的 Stream 类型，消息确认后不删除，可以重复读取。
 
+选型时先看业务需要什么：
+
+- 日志、埋点、数据同步这类数据量大、需要多个系统各自消费或者回溯重放的，选 Kafka；
+- 订单、支付等业务消息需要延迟消息、事务消息、失败重试和死信时，选 RocketMQ，可以少写很多自研的代码；
+- 消息量不大、路由规则复杂（按路由键或通配符分发到不同的队列）的，选 RabbitMQ。
+
+此外还要考虑团队的运维经验和已有的技术栈。同时维护多种消息队列的成本很高，已有的一种能满足需求时通常继续使用。
+
 Redis 也常被用作轻量的消息队列，但可靠性不如专门的消息队列：
 
 - Pub/Sub 不保存消息，发布时没有订阅者或者订阅者断线，消息就丢了，实测没有订阅者时 PUBLISH 返回 0；
@@ -265,6 +298,8 @@ acks=all 要配合 min.insync.replicas 使用，否则 ISR 只剩 Leader 一个
 
 消费者处理完消息后提交偏移量，下次从提交的位置继续读取。默认每 5 秒自动提交一次，已经处理但还没提交时崩溃，重启后就会重复消费。例如消费者读到 3 条扣款消息后被 kill -9，这时组的偏移量仍然为空；它在 session.timeout.ms（默认 45 秒）之后被移出消费者组，重新启动的消费者又收到了这 3 条消息（见[附录 A2](#a2-重复消费)）。因此接收端必须做到幂等，方法见 [接收端的可靠性](#接收端的可靠性)。反过来，如果先提交偏移量再处理，处理失败的消息就丢失了。
 
+消费者用线程池并行处理时也会遇到这个问题。提交偏移量 N 表示 N 之前的消息都已经处理完。例如一次取回了偏移量 100～109，线程已经处理完 105～109，100～104 还在处理中，这时提交 110，进程崩溃后 100～104 就不会再被消费。所以只能提交连续处理完的位置，这里是 100，代价是重启后 105～109 会被重复消费。
+
 ### 存储
 
 每个分区在磁盘上是一个目录，日志按大小切分成多个段（Segment），文件名是段中第一条消息的偏移量。例如把段的大小设为 1 MB（默认 1 GB），向一个只有一个分区的主题写入 3 万条 100 字节的消息后，目录中的文件和大小如下（Kafka 4.4 实测，省略了 .timeindex 等文件）：
@@ -421,3 +456,6 @@ $ redis-cli publish ch:orders 1024               # 没有订阅者
 - [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/)
+- [KafkaProducer（Kafka 4.3 Javadoc）](https://kafka.apache.org/43/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html)
+- [Queue Length Limit | RabbitMQ](https://www.rabbitmq.com/docs/maxlength)
+- [Memory and Disk Alarms | RabbitMQ](https://www.rabbitmq.com/docs/alarms)