Update 消息队列.md

补充 auto.offset.reset 对重启和新消费者组的影响,附录 A2 用 Kafka 4.4 重新实测。
This commit is contained in:
CyC2018
2026-11-04 22:39:58 +08:00
parent 6ae4884320
commit 1fe784ca36
+24 -5
View File
@@ -352,7 +352,7 @@ acks=all 要配合 min.insync.replicas(默认为 1)使用,否则 ISR 只
生产者没有收到确认就会重试,这可能让同一条消息写入两次。Kafka 3.0 起默认开启幂等生产者(3.0.0、3.1.0 因缺陷未生效):broker 给每个生产者分配一个 ID,生产者给发往每个分区的消息编上序号,broker 发现序号重复就丢弃。后文查看 .log 文件时,每个批次都带有 producerId: 1001 和 baseSequence 字段。幂等只保证单个分区内、生产者自身重试时不重复,应用自己重发的消息不能去重;要让写入多个分区的消息同时成功或失败,需要使用事务,消费者设置 isolation.level=read_committed 后只读取已提交事务中的消息。
消费者处理完消息后提交偏移量,下次从提交的位置继续读取。默认每 5 秒自动提交一次,已经处理但还没提交时崩溃,重启后就会重复消费。例如消费者读到 3 条扣款消息后被 kill -9,这时组的偏移量仍然为空;它在 session.timeout.ms(默认 45 秒)之后被移出消费者组,重新启动的消费者又收到了这 3 条消息(见[附录 A2](#a2-重复消费))。因此接收端必须做到幂等,方法见 [接收端的可靠性](#接收端的可靠性)。反过来,如果先提交偏移量再处理,处理失败的消息就丢失了。
消费者处理完消息后提交偏移量,下次从提交的位置继续读取。默认每 5 秒自动提交一次,已经处理但还没提交时崩溃,重启后可能重复消费。例如消费者读到 3 条扣款消息后被 kill -9,这时组还没有提交过偏移量;它在 session.timeout.ms(默认 45 秒)之后被移出消费者组,重新启动的消费者从哪里开始由 auto.offset.reset 决定:设为 earliest 时从最早的消息开始,又收到了这 3 条消息;默认的 latest 从末尾开始,这 3 条不会再被自动读到(Kafka 4.4 实测,见[附录 A2](#a2-重复消费))。新的消费者组第一次订阅,或者已提交的偏移量对应的消息已被删除时,也按这个配置决定。因此接收端必须做到幂等,方法见 [接收端的可靠性](#接收端的可靠性)。反过来,如果先提交偏移量再处理,处理失败的消息就丢失了。
消费者用线程池并行处理时也会遇到这个问题。提交偏移量 N 表示 N 之前的消息都已经处理完。例如一次取回了偏移量 100~109,线程已经处理完 105~109,100~104 还在处理中,这时提交 110,进程崩溃后 100~104 就不会再被消费。所以要关闭自动提交(它可能早于处理完成),只手动提交连续处理完的位置,这里是 100,代价是重启后 105~109 会被重复消费。
@@ -425,21 +425,40 @@ $ ...(同上) acks=all batch.size=0 linger.ms=0
### A2 重复消费
消费者组 billing 使用默认的自动提交,读到 3 条消息后立即 kill -9,45 秒后再启动一个同组的消费者:
Kafka 4.4.0,本地 3 个节点,JDK 21。每种 auto.offset.reset 各用一个新的主题(1 个分区)和新的消费者组:先启动消费者,等 8 秒让它完成分区分配、确定初始位置,再生产 3 条消息;消费者开启自动提交,为了保证被 kill -9 时还没有提交过偏移量,把提交间隔设为 60 秒,生产、消费和 kill -9 都在启动后约 10 秒内完成;50 秒后再启动一个同组的消费者。两组的第一个消费者都收到了 3 条消息,说明生产时它已经就绪:
```bash
kafka-console-consumer.sh --bootstrap-server $B --topic $T --group $G \
--command-property auto.offset.reset=latest --command-property auto.commit.interval.ms=60000
```
```text
== auto.offset.reset=latest
第一个消费者:
扣款 order:1024
扣款 order:1025
扣款 order:1026
(kill -9)
重新启动的消费者:
kill -9 后组的偏移量:
PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
0 - 3 -
50 秒后重新启动的消费者:
(没有收到消息)
== auto.offset.reset=earliest
第一个消费者:
扣款 order:1024
扣款 order:1025
扣款 order:1026
kill -9 后组的偏移量:
PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
0 - 3 -
50 秒后重新启动的消费者:
扣款 order:1024
扣款 order:1025
扣款 order:1026
Processed a total of 3 messages
```
CURRENT-OFFSET 为“-”表示组在这个分区上还没有提交过偏移量。
### A3 消费积压
主题 points-events 有 3 个分区、3 个副本。用 kafka-console-producer.sh 写入 3 万条带 key 的消息(key 为 997 个用户 ID),消费者组 points 用 kafka-console-consumer.sh --max-messages 3000 消费 3000 条后退出,再查看积压;之后同时启动 4 个同组的消费者,查看每个消费者分到的分区数: