diff --git a/notes/消息队列.md b/notes/消息队列.md index 2d38ea69..5e70c9f8 100644 --- a/notes/消息队列.md +++ b/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)