From 1fe784ca36eec605b5c7780eb3cf54895964a71e Mon Sep 17 00:00:00 2001 From: CyC2018 Date: Wed, 4 Nov 2026 22:39:58 +0800 Subject: [PATCH] =?UTF-8?q?Update=20=E6=B6=88=E6=81=AF=E9=98=9F=E5=88=97.m?= =?UTF-8?q?d?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 补充 auto.offset.reset 对重启和新消费者组的影响，附录 A2 用 Kafka 4.4 重新实测。 --- notes/消息队列.md | 29 ++++++++++++++++++++++++----- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/notes/消息队列.md b/notes/消息队列.md index 3f3edab6..38ed20ba 100644 --- a/notes/消息队列.md +++ b/notes/消息队列.md @@ -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 个同组的消费者，查看每个消费者分到的分区数：