第 5 章 Consumer 深入：把 poll() 拆成消费组协议的六步舞

目标读者：第 4 章已经搞清楚 Producer 如何发消息，现在要学「消息怎么被读出来、怎么不重不漏、怎么在多消费者之间分摊分区」的同学。
学完你会：闭着眼睛说出 subscribe + poll 循环的内部含义；分得清自动提交 / 同步提交 / 异步提交 / 手动按位点提交的四种姿势及踩坑点；能完整画出消费者与 Coordinator 的 6 步协议（FindCoordinator → JoinGroup → SyncGroup → Heartbeat → OffsetCommit → LeaveGroup）；能根据「重复消费」「漏消费」的现象反推根因；会用 seek / pause / resume 处理高级场景。

0. 导读：Consumer 比 Producer 复杂十倍

很多人以为：「Producer 是发，Consumer 是收，应该差不多简单」。错。Consumer 端有：

多个 Consumer 要协同（同组分摊分区，同消息只被一个成员看到）；
Offset 要持久化（崩溃后能续上）；
Rebalance 要协调（成员加入 / 离开时重新分配分区）；
心跳要稳（broker 通过心跳判活）；
业务处理时间不可控（poll 间隔过长会被踢出组）。

任何一个细节没处理好，就会出现「消息重复 / 消息丢失 / Rebalance 风暴 / 死循环 poll」中的至少一个。本章把这 5 件事一次讲透。

python

consumer.subscribe(["learn.05.orders"])

加入消费组，由 Group Coordinator 分配分区。
多个消费者订阅同一个 Topic 时自动分摊分区。
成员加入 / 离开会触发 Rebalance。
典型用法：业务消费组、流处理。

1.2 assign（手动指定）

python

from confluent_kafka import TopicPartition
consumer.assign([
    TopicPartition("learn.05.orders", 0),
    TopicPartition("learn.05.orders", 2),
])

不加入消费组，自己决定消费哪些分区。
没有 Rebalance、没有自动分摊、没有 Coordinator 协调。
Offset 仍然可以提交到 __consumer_offsets（只要配了 group.id），但更多场景下手动管。
典型用法：
- 数据修复 / 回放（指定从某个分区读）。
- Kafka Streams 内部（自己管理 task → partition 映射）。
- 某些自定义流量调度场景。

1.3 对比表

维度	subscribe	assign
加入消费组	✅	❌
Rebalance	✅ 自动	❌ 没有
分区分配策略	Range / RoundRobin / Sticky / CooperativeSticky	自己控
Offset 提交	推荐	可选 / 自管
多实例分摊	自动	自己保证不重复
灵活度	标准	极高
复杂度	简单	自己处理一切

⚠️ 不能混用：同一个 Consumer 实例 subscribe 后再 assign 会抛 IllegalStateException，反之亦然。

2. poll 循环：消费者的「心跳」

2.1 最小骨架

python

consumer.subscribe(["learn.05.orders"])
try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            handle_error(msg.error())
            continue
        process(msg)        # 业务逻辑
finally:
    consumer.close()        # 关键：触发同步 commit + LeaveGroup

2.2 poll() 在内部干了什么

consumer.poll(1.0)
  ├─ 如果消费组未加入 → JoinGroup + SyncGroup（首次或刚 Rebalance 完）
  ├─ 发送 Heartbeat（如果上次心跳超过 heartbeat.interval.ms）
  ├─ 提交 Offset（如果开了 enable.auto.commit 且到期）
  ├─ 从已分配分区拉取数据（Fetch 请求）
  │    - 缓存在本地 buffer
  │    - 解压 / 反序列化
  ├─ 返回一条消息（或 None 如果超时还没拉到）
  └─ 若 fetch 拿到一批，下次 poll 直接从 buffer 取，不再发 Fetch

关键认知：

poll() 不仅是「拉消息」，还是 Consumer 的「心跳节拍器」——心跳、自动提交、Rebalance 都依赖它被定时调用。
不能在两次 poll 之间做太长时间的处理（超过 max.poll.interval.ms 默认 5 分钟会被踢出组）。
单次 poll() 最多返回 max.poll.records（默认 500）条。

2.3 timeout 的语义

timeout	行为
`0`	立即返回，本地 buffer 没数据就回 `None`
`>0`	阻塞最多 N 秒等待新数据
`-1`	无限阻塞（不推荐，无法响应中断）

经验值：poll(1.0)——1 秒一个节拍，心跳够频繁，又能及时响应 SIGINT。

2.4 max.poll.records 的取舍

python

Consumer({"max.poll.records": 100})

小（100-500）：单次 poll 处理快，对 max.poll.interval.ms 友好；适合处理慢的业务。
大（500-2000）：单次拉的多，吞吐高；要求业务能在 max.poll.interval.ms 内处理完。

3. Offset 提交：4 种姿势 + 4 种坑

3.1 自动提交（默认）

python

Consumer({
    "enable.auto.commit": True,
    "auto.commit.interval.ms": 5000,
})

poll() 内部每 5 秒自动提交一次「上次 poll 拿到的最大 offset + 1」。
✅ 优点：零代码，开箱即用。
❌ 坑：「先消费后提交」陷阱：
- 消息已经从 poll 拿出来 → 业务还没处理完 → 5 秒到了，自动 commit
- 进程崩溃 → 重启后从 commit 之后开始 → 业务没处理完那段消息丢了（漏消费）。
- 或者：业务处理完 → 5 秒还没到 → 进程崩溃 → 重启从老 commit 开始 → 重复消费。
结论：自动提交只有在「消息丢了不要紧」场景下才能用（指标采样、可重放日志）。

3.2 同步提交

python

Consumer({"enable.auto.commit": False})

while True:
    msg = consumer.poll(1.0)
    if msg is None: continue
    process(msg)
    consumer.commit(message=msg, asynchronous=False)  # 阻塞直到 broker ack

✅ 准确：处理完一条提交一条，崩溃后只会重消费当前那一条。
❌ 慢：每次 commit = 一次 broker RPC，吞吐降低 80%+。
使用场景：消息少 + 极重要（财务、订单）。

3.3 异步提交

python

def cb(err, partitions):
    if err: print("commit failed:", err)

while True:
    msg = consumer.poll(1.0)
    if msg is None: continue
    process(msg)
    consumer.commit(message=msg, asynchronous=True, callback=cb)

✅ 快：不等 broker ack，立即继续。
❌ 失败时不会自动重试（重试可能让旧 offset 覆盖新 offset，反而错乱）。
典型组合：循环里异步 commit + finally 里同步 commit 兜底。

3.4 批次提交（最实用）

python

batch = []
while True:
    msg = consumer.poll(1.0)
    if msg is None:
        if batch:
            process_batch(batch)
            consumer.commit(asynchronous=False)
            batch = []
        continue
    batch.append(msg)
    if len(batch) >= 500:
        process_batch(batch)
        consumer.commit(asynchronous=False)
        batch = []

✅ 兼顾吞吐和准确性（每 500 条 commit 一次）。
✅ 减少与 broker 的 RPC 次数。
❌ 崩溃时最多重消费 500 条（业务必须幂等才安全）。
业界主流做法。

3.5 手动按位点提交（Exactly-Once 关键）

python

from confluent_kafka import TopicPartition

while True:
    msgs = consumer.consume(num_messages=100, timeout=1.0)
    for m in msgs:
        if m.error(): continue
        process(m)

    if msgs:
        last = max(msgs, key=lambda m: (m.partition(), m.offset()))
        # 注意：commit 的 offset 是「下一条要消费的位置」= last.offset() + 1
        consumer.commit(
            offsets=[TopicPartition(last.topic(), last.partition(), last.offset() + 1)],
            asynchronous=False,
        )

显式控制每个分区的提交位置。
EOS 场景必备：先把业务结果写下游 + 再提交 offset，原子性由事务保证。

3.6 「消息丢失」vs「重复消费」根因表

现象	提交方式	根因	修复
重复消费	自动提交	commit 间隔 > 处理时间，崩溃前已处理但未 commit	改手动批次提交
重复消费	异步提交	异步 commit 失败被忽略，下次仍从老位置	加同步兜底
漏消费	自动提交	消息从 poll 出来就计入 commit，业务没处理完就 commit 了	改手动「处理后提交」
漏消费	手动提交	业务异常但 catch 后没回滚 → 仍 commit	异常时不要 commit

核心原则：「消费 → 处理 → 提交」的顺序不能变，并且业务要幂等。

4. `__consumer_offsets`：Offset 的「家」

4.1 它是什么

一个特殊的内部 Topic，默认 50 个分区（offsets.topic.num.partitions）。
副本数由 offsets.topic.replication.factor 控制，生产环境 3。
Cleanup 策略 = compact（按 Key 压缩，只保留每个 Key 的最新值）。
存储所有消费组的 offset 提交记录，KRaft 模式下也是它。

4.2 Key 编码

每条 record 的 Key 大致结构：

type=offset_commit
  group="team-A"
  topic="learn.05.orders"
  partition=2

Value 大致结构：

{
  offset: 1234,
  metadata: "...",  // 可选，应用自定义
  commit_timestamp: 1713344000000
}

每个 (group, topic, partition) 三元组唯一，Compaction 保证只留最新位点。

4.3 怎么查看

用 kafka-console-consumer.sh + 专用 formatter：

bash

kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 \
  --topic __consumer_offsets \
  --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" \
  --from-beginning \
  --max-messages 10

输出：

[team-A,learn.05.orders,0]::OffsetAndMetadata(offset=120, leaderEpoch=Optional[3], metadata=, commitTimestamp=1713344000123, expireTimestamp=None)
[team-A,learn.05.orders,1]::OffsetAndMetadata(offset=205, leaderEpoch=Optional[3], metadata=, commitTimestamp=1713344000456)
[team-B,learn.05.orders,0]::OffsetAndMetadata(offset=99,  leaderEpoch=Optional[3], metadata=, commitTimestamp=1713344000789)

怎么定位某个 group 的分区：

partition_for_group = abs(hash("team-A")) % offsets.topic.num.partitions
                    = abs(hash("team-A")) % 50

同一个 group 的所有 offset 都在同一个分区，方便 Coordinator 集中加载。

4.4 删除 / 过期

默认 offsets 保留 7 天（offsets.retention.minutes=10080）。
消费组 7 天没提交过任何 offset → 被认为是 dead group → offsets 被 compaction 删除（写入 tombstone）。
重启消费组时如果发现 offsets 都没了 → 触发 auto.offset.reset 兜底逻辑。

5. auto.offset.reset：找不到 offset 时怎么办

python

Consumer({"auto.offset.reset": "earliest"})  # earliest / latest / none

取值	含义	典型场景
`earliest`	从分区最早消息开始	数据回放、首次上线想看历史
`latest`	从分区末尾开始（默认）	只关心新消息
`none`	报错 `NoOffsetForPartitionException`	EOS / 严格场景，让人工干预

⚠️ 生效条件：只有在「该消费组对该分区没有已提交 offset」时才生效。已经有 offset 的话直接从那里继续，与 auto.offset.reset 无关。

经典踩坑：「新业务上线，第一次启动 + latest → 历史数据全没消费 → 业务以为没数据」。修复：上线第一版用 earliest，确认追上后再切回 latest，或干脆永久 earliest（更安全）。

6. 心跳与超时四参数

session.timeout.ms        默认 45000（KIP-735 之后）
heartbeat.interval.ms     默认 3000
max.poll.interval.ms      默认 300000 (5 min)
group.instance.id         可选，配了后是「Static Membership」

6.1 各自作用

参数	控制什么
`heartbeat.interval.ms`	Consumer 后台线程多久发一次 Heartbeat
`session.timeout.ms`	Coordinator 多久收不到心跳就认为成员死亡
`max.poll.interval.ms`	两次 poll() 之间最长时间，超过认为 Consumer 卡死
`group.instance.id`	静态成员标识，重启不触发 Rebalance

6.2 关系图

        ┌────── heartbeat.interval.ms ───────┐ (3s)
        │                                    │
   poll │   后台心跳线程定期发 HB             │
        │ ◄──────────────────────────────►   │
        │                                    │
        ▼                                    ▼
  ┌──────────────────────────────────────────────────┐
  │  session.timeout.ms (45s)                        │
  │  Coordinator 这么久没收到 HB → 踢出成员          │
  └──────────────────────────────────────────────────┘

        ▲
        │
   poll │   max.poll.interval.ms (5 min)
        │  两次 poll 之间超过这个时间 → 踢出（业务卡死）
        ▼
  ┌──────────────────────────────────────────────────┐
  │  Consumer 主线程要保证持续 poll                  │
  └──────────────────────────────────────────────────┘

6.3 经典关系

heartbeat.interval.ms < session.timeout.ms     （一般 1/3 关系）
session.timeout.ms ≤ group.max.session.timeout.ms（broker 端约束）
max.poll.interval.ms ≥ 业务最长单次处理时间 × max.poll.records

6.4 常见踩坑

6.4.1 `max.poll.interval.ms` 引发 Rebalance 风暴

业务处理一条慢（比如调用外部 API 5 秒），max.poll.records=500 → 单次 poll 后处理 500*5=2500 秒 → 远超 max.poll.interval.ms=300 秒 → 被踢 → 触发 Rebalance → 别的成员接手又卡 → 死循环。

修复：

降低 max.poll.records（如 100）。
提高 max.poll.interval.ms（如 1800000 = 30 分钟）。
业务能异步化 / 批量处理就异步 / 批量。
用 pause() 暂停拉新数据再处理慢任务（见第 8 节）。

6.4.2 短暂网络抖动触发 Rebalance

session.timeout.ms 小（10 秒）+ 偶尔 GC 暂停 5 秒 → 可能错过 1-2 次心跳 → 被踢。

修复：调大 session.timeout.ms 到 30-60 秒（同时 broker 端 group.max.session.timeout.ms 要够大）。

6.4.3 滚动重启触发 N 次 Rebalance

3 个消费者轮流重启，每次重启都触发一次 Rebalance（共 6 次：离开 + 加入）。

修复：用 Static Membership：

python

Consumer({
    "group.instance.id": "consumer-1",   # 每个实例一个固定 id
    "session.timeout.ms": 60000,
})

配了 group.instance.id 后，Consumer 重启 保留成员资格（broker 等 session.timeout 内回来），不触发 Rebalance。
滚动重启从 6 次 Rebalance 减少到 0 次。
⚠️ 必须保证不同实例的 group.instance.id 不冲突，否则会互相踢。

7. Consumer 与 Coordinator 的 6 步协议

「Consumer 怎么找到自己的 Coordinator？怎么加入组？怎么收到分区分配结果？」——一切都是 6 个 RPC 编排出来的。

7.1 6 步详解

Step 1. FindCoordinator
  Consumer ──► 任意 Broker
                     ↓ 计算 partition_for_group = hash(group.id) % 50
                     ↓ 查 __consumer_offsets 那个 partition 的 Leader Broker
              ◄── coordinator_broker_id

Step 2. JoinGroup
  Consumer ──► Coordinator
       (member.id, group.id, subscribed topics, supported assignors)
                     ↓ 等所有同组成员都到齐 (rebalance.timeout.ms)
                     ↓ 选一个成员当「Leader」（第一个到的）
              ◄── (你的 member.id, leader_id, members 列表, 选定的 assignor)

Step 3. SyncGroup
  Leader Consumer:
    根据成员列表 + assignor 算分区分配方案 ──► Coordinator
  其他 Consumer:
    空请求 ──► Coordinator (只为同步等结果)
                     ↓ Coordinator 把方案存到 __consumer_offsets
              ◄── 你被分配到的分区列表

Step 4. Heartbeat (循环)
  Consumer ──► Coordinator (每 heartbeat.interval.ms)
              ◄── (OK 或 REBALANCE_IN_PROGRESS)
       如果收到 REBALANCE_IN_PROGRESS → 主动重新 JoinGroup

Step 5. OffsetCommit
  Consumer ──► Coordinator (按提交策略)
              ◄── ack 或错误

Step 6. LeaveGroup（优雅退出时）
  Consumer ──► Coordinator
              ◄── 立即触发 Rebalance（不等 session.timeout）

7.2 时序图

7.3 4 种分区分配策略（assignor）

策略	行为	适用
Range（默认）	每个 Topic 内按字典序分段：C1 拿前一段，C2 拿后一段	简单
RoundRobin	所有 (Topic, Partition) 联合后轮询分配	多 Topic 场景下更均匀
Sticky	在尽量保持原有分配的前提下做最小改动	减少 Rebalance 时的状态丢失
CooperativeSticky	在 Sticky 基础上做增量再平衡（Eager Rebalance 的「Stop-the-World」之痛被消除）	2.4+ 推荐

⚠️ CooperativeSticky 是 2025 年的事实标准：Rebalance 时不再「全员暂停 → 全部重新分配」，而是 「只有受影响的分区移交，其它继续工作」，业务停顿降到接近 0。

confluent-kafka 配置：

python

Consumer({"partition.assignment.strategy": "cooperative-sticky"})

8. 高级 API：seek / pause / resume

8.1 seek：跳到任意 offset

python

from confluent_kafka import TopicPartition

# 跳到指定 offset
consumer.seek(TopicPartition("learn.05.orders", 0, 1234))

# 跳到分区开头
consumer.seek_to_beginning(consumer.assignment())

# 跳到分区末尾
consumer.seek_to_end(consumer.assignment())

# 按时间戳跳
tps = [TopicPartition("learn.05.orders", p, int(time.time()*1000) - 3600000)
       for p in range(6)]
offsets = consumer.offsets_for_times(tps)
for tp in offsets:
    consumer.seek(tp)

⚠️ seek 必须在 partition 已分配后才能调用——刚 subscribe 后第一次 poll() 才会触发分配，所以一般要在 on_assign 回调里 seek。

python

def on_assign(c, partitions):
    for p in partitions:
        p.offset = 0          # 强制从头
    c.assign(partitions)

consumer.subscribe(["learn.05.orders"], on_assign=on_assign)

8.2 pause / resume：暂停某些分区

python

# 业务下游处理不过来，暂停某些分区
consumer.pause([TopicPartition("learn.05.orders", 2)])

# 恢复
consumer.resume([TopicPartition("learn.05.orders", 2)])

典型场景：

8.2.1 处理慢任务时避免 Rebalance

业务突然要做一个 30 分钟的耗时操作（导出报表）：

python

consumer.pause(consumer.assignment())   # 暂停所有
# 心跳照常发，不会触发 max.poll.interval.ms
do_long_task()
# 业务期间循环 poll() 触发心跳
consumer.resume(consumer.assignment())

⚠️ 即使 pause 了，仍要继续调 poll()！否则心跳停 → 仍会被踢。pause 只是让 poll 不返回那个分区的数据。

8.2.2 背压控制

下游 DB 压力大时，pause 几个分区让 lag 升高，等下游缓过来再 resume。

8.3 commit 指定位点

python

from confluent_kafka import TopicPartition, OFFSET_BEGINNING, OFFSET_END

# 强制把 group 的 offset 重置到分区开头（消费组停止后才能这样做）
consumer.commit(offsets=[
    TopicPartition("learn.05.orders", 0, OFFSET_BEGINNING),
    TopicPartition("learn.05.orders", 1, OFFSET_BEGINNING),
])

等价于 kafka-consumer-groups.sh --reset-offsets --to-earliest --execute，但用代码实现。

9. 与其他 MQ 对比

维度	Kafka Consumer	RabbitMQ Consumer	RocketMQ Consumer
拉/推	Pull（poll）	Push（broker 推 + ACK）	Pull（PullConsumer）/ Push 包装
偏移管理	`__consumer_offsets` 内部 topic	broker 内存 + Queue ack	broker 端 OffsetStore
消费组	显式 group.id，分摊分区	多个 consumer 公共队列时共享，无组概念	ConsumerGroup（与 Kafka 类似）
Rebalance	Range/RoundRobin/Sticky/CooperativeSticky	无（由 broker 自动派发）	AllocateMessageQueueAveragely / Hash 等
重复消费	提交策略导致	broker 重投递（autoAck=false 时）	消费失败重投，最多 16 次
顺序消费	单分区有序	单队列有序	顺序消息（按 MessageQueue）
流控	pause / resume / `max.poll.records`	Prefetch（QoS）	PullThresholdForQueue
消费回放	seek + reset-offsets 任意点	不能（消息一旦 ack 就删）	支持按 offset / 时间回溯

📌 核心差异：Kafka 的「pull + offset」给了消费者完全的回放能力——RabbitMQ 一旦 ack 消息就没了，Kafka 可以从 7 天前重头消费。这就是「Kafka = 可重放的分布式日志」的具体体现。

10. 调优清单

Consumer 调优 8 步走
1) 选 subscribe 还是 assign
   - 标准业务 → subscribe
   - 数据修复 / 自管 → assign
2) 选分区分配策略
   - 2025 年首选 cooperative-sticky
3) 选 Offset 提交策略
   - 业务幂等 + 高吞吐 → 批次同步提交（每 N 条）
   - EOS → 手动按位点 + 事务 Producer
   - 可丢 → 自动提交
4) 心跳与超时
   - heartbeat.interval.ms = session.timeout.ms / 3
   - max.poll.interval.ms ≥ 业务最长处理时间
   - 滚动重启场景 → group.instance.id (Static Membership)
5) 调 max.poll.records
   - 处理慢 → 调小（100）
   - 处理快 → 调大（1000-2000）
6) auto.offset.reset
   - 新业务首次上线 → earliest
   - 只关心新数据 → latest
   - EOS → none
7) 业务幂等
   - 用业务主键 + 去重表 / Redis SETNX / 数据库唯一约束
8) 监控关键指标
   - records-lag-max         （最大 lag，最重要）
   - records-consumed-rate   （消费速率）
   - rebalance-rate          （Rebalance 频率，>1/小时就要排查）
   - commit-latency-avg/p99  （commit 耗时）
   - poll-idle-ratio-avg     （poll 空闲比例，低说明业务太慢）

11. 实战脚本一览

本章配套目录 05_consumer/：
init.sh —— 创建本章 Topic、灌一些消息。
code/auto_commit_consumer.py —— 默认自动提交 + 演示「未处理完就 commit」的漏消费。
code/manual_commit_consumer.py —— 手动批次提交 + 同步 commit 兜底。
code/seek_demo.py —— on_assign 回调里 seek 到指定 offset / 时间戳。
code/pause_resume_demo.py —— 模拟下游慢、用 pause/resume 做背压。
demo.html —— 浏览器交互：可加入 / 离开 Consumer，自动重绘分区分配；Offset 提交动画；心跳超时演示。

12. 小结

Consumer 深入
├─ 订阅
│  ├─ subscribe → 入组 + Rebalance + 自动分摊
│  └─ assign    → 自管，无 Rebalance
├─ poll 循环
│  ├─ 是「心跳节拍器」，必须持续调
│  ├─ 内部：JoinGroup / Heartbeat / Fetch / OffsetCommit
│  └─ 控制：max.poll.records / poll(timeout)
├─ Offset 提交（4 种）
│  ├─ 自动：简单，可能丢/重
│  ├─ 同步：准但慢
│  ├─ 异步：快但失败不会重试
│  └─ 批次（业界主流）：幂等业务 + 每 N 条同步 commit
├─ __consumer_offsets
│  ├─ 50 分区，compact 策略
│  ├─ Key = (group, topic, partition)
│  └─ partition_for_group = hash(group) % 50
├─ auto.offset.reset
│  └─ earliest / latest / none，仅在无 offset 时生效
├─ 心跳 4 参数
│  ├─ heartbeat.interval.ms (3s)
│  ├─ session.timeout.ms (45s)
│  ├─ max.poll.interval.ms (5min)
│  └─ group.instance.id (Static Membership)
├─ Coordinator 6 步
│  └─ FindCoordinator → JoinGroup → SyncGroup → Heartbeat → OffsetCommit → LeaveGroup
├─ 分区分配策略
│  └─ Range / RoundRobin / Sticky / CooperativeSticky (2025 默认推荐)
└─ 高级 API
   ├─ seek / seek_to_beginning / seek_to_end / offsets_for_times
   └─ pause / resume （配合心跳保活）

13. 面试高频题（6 题）

Q1：自动提交 vs 手动提交，怎么选？分别会怎么丢/重消费？

考察点：消费语义最高频题。

答案：

自动提交 (enable.auto.commit=true, auto.commit.interval.ms=5000)：
- 每 5 秒在 poll() 内部自动提交「上次 poll 拿到的最大 offset + 1」。
- 漏消费：消息从 poll 拿出来还没处理完，5 秒到了 → 自动 commit → 进程崩溃 → 重启从 commit 之后开始 → 没处理完那批消息丢了。
- 重复消费：业务处理完了但 commit 还没到 → 进程崩溃 → 重启从老位置开始 → 重复消费。
- 适用：消息可丢失场景（指标采样、可重放日志）。
同步提交 (commit(asynchronous=False))：
- 处理一条 commit 一条，崩溃只重消费当前一条。
- 吞吐降低 80%+，每条都阻塞等 broker。
- 适用：消息少 + 极重要。
异步提交 (commit(asynchronous=True, callback=cb))：
- 不阻塞，吞吐高。
- 失败不会自动重试（避免新 offset 被旧 offset 覆盖）。
- 典型组合：循环里异步 commit + finally 同步兜底。
批次提交（业界主流）：
- 每 100-1000 条消息处理完后同步 commit 一次。
- 兼顾吞吐和准确性，崩溃时最多重消费一个 batch。
- 要求业务幂等（业务主键去重 / Redis SETNX / DB 唯一约束）。
EOS 场景：手动按位点提交 + 事务 Producer，把「写下游 + commit offset」放进同一个 Kafka 事务（sendOffsetsToTransaction）。
核心原则：消费 → 处理 → 提交，顺序不能换；异常时不要 commit；业务必须幂等。

加分项：提到 enable.auto.commit=true 时 commit 时机是 poll() 调用时检查、不是 5 秒定时；isolation.level=read_committed 时 Consumer 只会消费已 commit 的事务消息。

Q2：`max.poll.interval.ms` 和 `session.timeout.ms` 的区别？踩坑？

考察点：心跳与 Rebalance 的本质。

答案：

session.timeout.ms（默认 45s）：Coordinator 多久没收到心跳就判定成员死亡。
- 由后台心跳线程负责（每 heartbeat.interval.ms 一次，默认 3s）。
- 与业务无关，纯网络 / 进程存活检查。
max.poll.interval.ms（默认 5min）：两次 poll() 之间最长间隔，超过认为 Consumer 业务卡死。
- 由 Coordinator 检查：每次 poll() 的时候 client 会上报「我还活着」的时间戳。
- 主要防范业务一直处理不完 → 心跳虽然在发，但业务永远不再 poll 新消息。
典型踩坑：
- max.poll.interval.ms 引发 Rebalance 风暴：处理一条慢（5s）+ max.poll.records=500 → 单批 2500s > 5min → 被踢 → 别人接手又卡 → 死循环。
- 修复：调小 max.poll.records（如 100），调大 max.poll.interval.ms（如 30 分钟），慢任务用 pause() + 持续 poll 保心跳。
静态成员 (group.instance.id) 让 Consumer 重启时保留资格（broker 等 session.timeout 内回来），把滚动重启的 N 次 Rebalance 降到 0。
关系：
- heartbeat.interval.ms < session.timeout.ms（一般 1/3）。
- max.poll.interval.ms ≥ 业务最长处理时间 × max.poll.records。

加分项：提到 KIP-735（Kafka 3.0+）把默认 session.timeout.ms 从 10s 调到 45s，正是为了减少误判 / Rebalance 风暴。

Q3：Consumer 与 Coordinator 的 6 步协议是什么？Rebalance 怎么触发？

考察点：消费组核心协议。

答案：

6 步：
1. FindCoordinator：Consumer 向任意 broker 询问，broker 算 hash(group.id) % 50 找到 __consumer_offsets 那个 partition 的 Leader → 它就是 Coordinator。
2. JoinGroup：Consumer 上报自己的 member.id / 订阅 / 支持的 assignor。Coordinator 等所有同组成员到齐（rebalance.timeout.ms），选第一个为 Leader。
3. SyncGroup：Leader 算分区分配方案，发给 Coordinator；其它成员发空 SyncGroup 等结果；Coordinator 把方案存到 __consumer_offsets 再下发。
4. Heartbeat：每 heartbeat.interval.ms 一次心跳。Coordinator 收不到就触发 Rebalance；返回 REBALANCE_IN_PROGRESS 告诉客户端要重新 JoinGroup。
5. OffsetCommit：按提交策略写 offset。
6. LeaveGroup：优雅退出时主动通知，立即触发 Rebalance，不等 session.timeout。
Rebalance 触发条件：
- 成员加入（新 Consumer 启动并订阅）。
- 成员离开（正常 LeaveGroup 或 session timeout / max.poll.interval 超时）。
- 订阅的 Topic 分区数变化（加分区）。
- 元数据变化（订阅了新 Topic 等）。
分区分配策略（4 种）：
- Range（默认）：按 Topic 内字典序分段。
- RoundRobin：所有 (Topic, Partition) 联合后轮询。
- Sticky：尽量保持原有分配。
- CooperativeSticky（2.4+ 推荐）：增量再平衡，无 Stop-the-World。
Eager vs Cooperative：
- Eager（前 3 种）：所有成员同时停掉所有分区 → 重新算 → 重新拿。业务停顿大。
- Cooperative：只有受影响的分区移交，其它分区继续工作。停顿接近 0。

加分项：提到 KRaft 模式下 Coordinator 行为没变（仍然是某个 broker 持有 __consumer_offsets 的 leader 角色）；group.instance.id 配了之后是 Static Membership，正常重启不触发 Rebalance。

Q4：`__consumer_offsets` 是什么？为什么用 compact 策略？

考察点：内部 Topic 与存储模型。

答案：

是什么：Kafka 内部 Topic，存所有消费组的 offset 提交记录。
配置：
- 默认 50 个分区（offsets.topic.num.partitions）。
- 副本数 3（offsets.topic.replication.factor）。
- cleanup.policy = compact。
Key 编码：(group_id, topic, partition) 三元组。
Value：{ offset, leaderEpoch, metadata, commit_timestamp }。
为什么 compact：
- 同一个 (group, topic, partition) 会反复提交，每次产生一条 record。
- 普通 delete 策略需要按时间清理，时间不到就一直堆积。
- compact 策略只保留每个 Key 的最新 record，自然把「这个组在该分区的最新 offset」留下，旧的清掉。
- 完美匹配「offset = 一份按 key 持续覆盖的状态」的语义。
找到一个 group 的位置：partition = hash(group_id) % 50，所以同一个 group 的所有 offset 都在同一个分区，方便 Coordinator 集中加载（Coordinator 就是该分区的 Leader）。

怎么查看：

bash

kafka-console-consumer.sh --bootstrap-server xxx \
  --topic __consumer_offsets \
  --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" \
  --from-beginning

过期：默认 7 天（offsets.retention.minutes）。死组的 offsets 写 tombstone，下次 compaction 物理删除。

加分项：提到 __transaction_state 也是 compact 策略，存事务状态机；KRaft 的 __cluster_metadata 是 Raft 日志，不是 compact，但有快照机制压缩。

Q5：CooperativeSticky 和 Sticky 的区别？为什么是 2025 年默认推荐？

考察点：Rebalance 演进、增量再平衡。

答案：

Eager Rebalance（Range / RoundRobin / Sticky 都是这种）：
- Rebalance 时，所有成员同时调用 onPartitionsRevoked（释放所有分区）。
- 然后 JoinGroup + SyncGroup 重新分配。
- 最后 onPartitionsAssigned（重新拿）。
- 整个过程所有 Consumer 都停止消费（Stop-the-World），业务停顿可达数秒。
CooperativeSticky（增量再平衡，KIP-429，2.4+）：
- 第一次 Rebalance：算出新方案，告诉每个成员「你要释放哪些分区」（不是全部）。
- 没受影响的分区继续消费（业务不停）。
- 受影响的分区被释放后，第二次 Rebalance 把它们交给新主人。
- 结果：业务停顿接近 0，只有「真的要换主人」的分区有短暂停顿。

配置：

python

Consumer({"partition.assignment.strategy": "cooperative-sticky"})

优势：
- 滚动重启 / 弹性扩缩容时，业务几乎无感。
- Stream 处理 / 状态丰富的业务受益最大（避免反复重建状态）。
代价：
- Rebalance 走两轮协议，整体 Rebalance 时间略长。
- 客户端实现更复杂（broker 端兼容，但客户端要支持）。
2025 年事实标准：所有主流客户端（Java、librdkafka、Sarama 等）都已默认或推荐 CooperativeSticky。Kafka Streams 默认就是它。

加分项：提到迁移要小心「同组内不能混用 Eager 和 Cooperative」（broker 会拒绝），需要先全局升级一次；group.instance.id 的 Static Membership 与 Cooperative 是正交特性，可以叠加。

Q6：怎么实现「不重不漏」消费？

考察点：EOS 完整方案。

答案：

「不漏」的关键：业务处理完成后才能 commit offset。所以关掉自动提交，自己控时机。
「不重」的关键：业务幂等。两条路：
- 业务侧幂等：用业务唯一键（订单号 / 事件 ID）做去重（DB 唯一约束 / Redis SETNX / 去重表）。崩溃后即使重消费，也不会产生重复结果。
- Kafka 端 EOS：isolation.level=read_committed + 上游事务 Producer + sendOffsetsToTransaction。

业务幂等方案（最常用）：

python

while True:
    msgs = consumer.consume(num_messages=500, timeout=1.0)
    for m in msgs:
        if m.error(): continue
        if dedup.exists(m.key()): continue
        process(m)
        dedup.add(m.key())
    consumer.commit(asynchronous=False)

同步 commit 兜底。
业务异常时不 commit。
即使重消费，dedup 表挡住重复。

Kafka EOS 方案（第 13 章细讲）：
- 上游 Producer 配 transactional.id 开事务。
- Consumer 配 isolation.level=read_committed，只看已 commit 的事务消息。
- 处理逻辑里把「写下游 + commit consumer offset」放进同一个事务：
  java
```
producer.beginTransaction();
for (msg : msgs) producer.send(downstream_record);
producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();
```
  1
  2
  3
  4
- 整个过程要么全成功（offset 提交 + 下游写入），要么全失败（事务回滚），实现严格 EOS。
取舍：
- 业务幂等方案：吞吐高、对外部依赖（去重表）依赖大。
- Kafka EOS 方案：吞吐略低（事务开销 ~10%-20%），无外部依赖。
错误的常见做法：
- 自动提交 + 业务里 try/catch 默认成功 → 漏消费。
- 异步 commit 失败不处理 → 重复消费。
- 没幂等做兜底 → 任何方案都不可靠。

加分项：提到 Kafka Streams 的 processing.guarantee=exactly_once_v2 就是上面 EOS 方案的封装；与 RocketMQ 半消息事务对比：RocketMQ 是「先发半消息 → 业务确认 → 提交 / 回滚」，Kafka 是「上游事务直接覆盖 send + offset commit」，两者哲学不同。

下一章 06_topic_design.md 我们会回到「设计」层面：分区数怎么选、副本数怎么选、Key 怎么设计、何时不能加分区只能重建 Topic——把消费组的能力上限和 Topic 设计联动起来。

🎬 可视化演示

演示加载缓慢或样式异常？点此在新标签页打开 ↗

💻 示例代码

auto_commit_consumer.pymanual_commit_consumer.pypause_resume_demo.pyseek_demo.py

python

"""
第 5 章 - 自动提交 Consumer + 漏消费演示
=================================

默认 enable.auto.commit=True 时，poll() 内部按 auto.commit.interval.ms
定期提交「上次 poll 拿到的最大 offset + 1」。

风险：
  - 漏消费：消息从 poll 出来 → 5s 到 → 自动 commit → 进程崩溃
            → 重启从 commit 之后开始 → 丢了那批没处理完的消息

本脚本演示这个陷阱：
  - 业务故意 sleep(2s) 处理慢
  - 当处理完 2 条后用 sys.exit(1) 模拟崩溃
  - 重启脚本会发现「丢了第 3、4 条」（如果 commit 已经发生）

运行：
    bash ../init.sh
    python auto_commit_consumer.py            # 第一次跑会"崩溃"
    python auto_commit_consumer.py            # 第二次跑看 lag

正确做法见 manual_commit_consumer.py。
"""

from __future__ import annotations

import os
import sys
import time

from confluent_kafka import Consumer, KafkaError

BOOTSTRAP = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1:9092"
TOPIC = "learn.05.orders"
GROUP = "ch5-auto-commit-demo"
CRASH_AFTER = int(os.environ.get("CRASH_AFTER", "2"))
SLOW_PROCESS_MS = int(os.environ.get("SLOW_PROCESS_MS", "2000"))


def main() -> None:
    consumer = Consumer(
        {
            "bootstrap.servers": BOOTSTRAP,
            "group.id": GROUP,
            "client.id": f"auto-commit-{os.getpid()}",
            "auto.offset.reset": "earliest",
            "enable.auto.commit": True,
            "auto.commit.interval.ms": 1000,  # 1 秒就 commit，方便复现
            "session.timeout.ms": 10000,
            "max.poll.interval.ms": 60000,
        }
    )
    consumer.subscribe([TOPIC])
    print(f"[auto-commit-demo] group={GROUP}, will crash after {CRASH_AFTER} msgs")

    processed = 0
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                print("  ... no message in 1s, lag may be 0")
                continue
            if msg.error():
                if msg.error().code() != KafkaError._PARTITION_EOF:
                    print(f"  err: {msg.error()}")
                continue

            print(
                f"  [{processed + 1:>3}] received P{msg.partition()}@{msg.offset()} "
                f"key={msg.key().decode() if msg.key() else None}"
            )

            time.sleep(SLOW_PROCESS_MS / 1000)
            processed += 1

            print(f"  ✓ processed (took {SLOW_PROCESS_MS}ms)")

            if processed >= CRASH_AFTER:
                print("\n💥 模拟崩溃！消息可能已被 auto-commit 但没处理完")
                print("   重启脚本看下一次能拿到多少 lag")
                # 故意不调 consumer.close()，模拟崩溃
                os._exit(1)
    finally:
        consumer.close()


if __name__ == "__main__":
    main()

python

"""
第 5 章 - 手动批次提交 Consumer (业界主流写法)
==================================================

特征：
- enable.auto.commit=False
- 处理完一批后同步 commit
- 业务异常时不 commit (等下次重消费)
- finally 里同步 commit 兜底

要求：业务必须幂等（用业务主键去重 / DB 唯一约束 / Redis SETNX）。

运行：
    bash ../init.sh
    python manual_commit_consumer.py
"""

from __future__ import annotations

import json
import os
import signal
import sys
import time

from confluent_kafka import Consumer, KafkaError

BOOTSTRAP = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1:9092"
TOPIC = "learn.05.orders"
GROUP = "ch5-manual-commit-demo"
BATCH_SIZE = 50

running = True


def stop(*_):
    global running
    print("\n[stop] received signal, exiting after final commit")
    running = False


signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)

seen_keys = set()


def process(msg) -> bool:
    try:
        payload = json.loads(msg.value())
        key = f"{msg.topic()}-{payload['order_id']}"
        if key in seen_keys:
            print(f"  [DUP] already processed {key}, skip")
            return True
        seen_keys.add(key)
        time.sleep(0.005)
        return True
    except Exception as e:
        print(f"  [ERR] processing failed: {e} (offset={msg.offset()})")
        return False


def flush(consumer, batch):
    consumer.commit(asynchronous=False)
    last = batch[-1]
    print(
        f"  [commit] {len(batch)} msgs, last P{last.partition()}@{last.offset()} "
        f"(committed offset={last.offset() + 1})"
    )


def main() -> None:
    consumer = Consumer(
        {
            "bootstrap.servers": BOOTSTRAP,
            "group.id": GROUP,
            "client.id": f"manual-commit-{os.getpid()}",
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
            "session.timeout.ms": 30000,
            "max.poll.interval.ms": 300000,
            "partition.assignment.strategy": "cooperative-sticky",
        }
    )
    consumer.subscribe([TOPIC])
    print(f"[manual-commit-demo] group={GROUP}, batch_size={BATCH_SIZE}")

    batch = []
    total = 0
    failed = 0

    try:
        while running:
            msg = consumer.poll(1.0)
            if msg is None:
                if batch:
                    flush(consumer, batch)
                    total += len(batch)
                    batch = []
                continue
            if msg.error():
                if msg.error().code() != KafkaError._PARTITION_EOF:
                    print(f"  poll err: {msg.error()}")
                continue

            ok = process(msg)
            if ok:
                batch.append(msg)
            else:
                failed += 1

            if len(batch) >= BATCH_SIZE:
                flush(consumer, batch)
                total += len(batch)
                batch = []
    finally:
        if batch:
            flush(consumer, batch)
            total += len(batch)
        print(f"\n[done] processed={total}, failed={failed}")
        consumer.close()


if __name__ == "__main__":
    main()

python

"""
第 5 章 - pause / resume 演示
==================================================

场景：模拟下游 DB 写入压力大，对部分分区做背压。

要点：
- pause(partitions) 后这些分区不再返回数据
- 但仍然要持续 poll() 触发心跳，否则会被踢出组
- 等下游缓过来再 resume

业务还有一个高频用法：处理一个长任务（如导出报表）时
pause 全部分区，主线程持续 poll() 保心跳，处理完再 resume。

运行：
    bash ../init.sh
    python pause_resume_demo.py
"""

from __future__ import annotations

import os
import signal
import sys
import time

from confluent_kafka import Consumer, KafkaError

BOOTSTRAP = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1:9092"
TOPIC = "learn.05.orders"
GROUP = f"ch5-pause-resume-{int(time.time())}"

running = True


def stop(*_):
    global running
    print("\n[stop] received signal")
    running = False


signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)


class FakeDB:
    """模拟下游 DB：每 5 秒进入一次 30 秒的 backpressure 状态"""

    def __init__(self):
        self._start = time.time()

    def is_busy(self) -> bool:
        elapsed = (time.time() - self._start) % 35
        return 5 < elapsed < 15

    def write(self, msg) -> None:
        time.sleep(0.005)


def main() -> None:
    consumer = Consumer(
        {
            "bootstrap.servers": BOOTSTRAP,
            "group.id": GROUP,
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
            "session.timeout.ms": 30000,
            "max.poll.interval.ms": 300000,
        }
    )
    consumer.subscribe([TOPIC])
    db = FakeDB()
    paused = False
    processed = 0

    try:
        while running:
            now_busy = db.is_busy()

            if now_busy and not paused:
                assignment = consumer.assignment()
                if assignment:
                    consumer.pause(assignment)
                    paused = True
                    print(f"[pause] DB busy, paused {len(assignment)} partitions; will keep polling for heartbeat")
            elif not now_busy and paused:
                assignment = consumer.assignment()
                if assignment:
                    consumer.resume(assignment)
                    paused = False
                    print(f"[resume] DB idle, resumed {len(assignment)} partitions")

            msg = consumer.poll(1.0)
            if msg is None:
                if paused:
                    print("  [paused] heartbeat-only poll, no msg returned (expected)")
                continue
            if msg.error():
                if msg.error().code() != KafkaError._PARTITION_EOF:
                    print(f"  err: {msg.error()}")
                continue

            db.write(msg)
            processed += 1
            if processed % 20 == 0:
                consumer.commit(asynchronous=False)
                print(f"  [progress] {processed} msgs processed and committed")
    finally:
        consumer.close()
        print(f"\n[done] processed={processed}")


if __name__ == "__main__":
    main()

python

"""
第 5 章 - seek / seek_to_beginning / offsets_for_times 演示
==================================================

3 种典型用法：
  1. seek_to_beginning：在 on_assign 回调里强制从头消费
  2. seek 到指定 offset
  3. offsets_for_times：按时间戳跳转

注意：seek 必须在 partition 已分配后才能调用，所以一般写在 on_assign 回调里。

运行：
    bash ../init.sh
    python seek_demo.py beginning          # 从头
    python seek_demo.py offset 5           # 跳到 offset 5
    python seek_demo.py time 60            # 跳到 60 秒前
"""

from __future__ import annotations

import sys
import time

from confluent_kafka import Consumer, KafkaError, TopicPartition

BOOTSTRAP = "127.0.0.1:9092"
TOPIC = "learn.05.seek_demo"
GROUP = f"ch5-seek-{int(time.time())}"

mode = sys.argv[1] if len(sys.argv) > 1 else "beginning"
arg = sys.argv[2] if len(sys.argv) > 2 else "0"


def make_on_assign(mode: str, arg: str):
    """返回 on_assign 回调，根据 mode 决定 seek 策略。"""

    def on_assign(consumer, partitions):
        print(f"[on_assign] mode={mode}, partitions={[(p.topic, p.partition) for p in partitions]}")

        if mode == "beginning":
            for p in partitions:
                p.offset = 0
            consumer.assign(partitions)

        elif mode == "offset":
            target = int(arg)
            for p in partitions:
                p.offset = target
            consumer.assign(partitions)
            print(f"  -> seek to offset {target} on every partition")

        elif mode == "time":
            seconds_ago = int(arg)
            ts = int((time.time() - seconds_ago) * 1000)
            tps_with_ts = [TopicPartition(p.topic, p.partition, ts) for p in partitions]
            offsets = consumer.offsets_for_times(tps_with_ts, timeout=10)
            for o in offsets:
                print(f"  P{o.partition} @ ts={ts} -> offset={o.offset}")
            consumer.assign(offsets)

        else:
            consumer.assign(partitions)
            print("  no seek, using committed/auto.offset.reset")

    return on_assign


def main() -> None:
    consumer = Consumer(
        {
            "bootstrap.servers": BOOTSTRAP,
            "group.id": GROUP,
            "auto.offset.reset": "latest",
            "enable.auto.commit": False,
        }
    )
    consumer.subscribe([TOPIC], on_assign=make_on_assign(mode, arg))

    deadline = time.time() + 8
    consumed = 0
    while time.time() < deadline:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() != KafkaError._PARTITION_EOF:
                print(f"err: {msg.error()}")
            continue
        consumed += 1
        print(f"[{consumed:>3}] P{msg.partition()}@{msg.offset()} value={msg.value().decode()}")

    print(f"\nconsumed {consumed} msgs in 8s")
    consumer.close()


if __name__ == "__main__":
    main()

auto_commit_consumer.py ↗ · manual_commit_consumer.py ↗ · pause_resume_demo.py ↗ · seek_demo.py ↗

第 5 章 Consumer 深入：把 poll() 拆成消费组协议的六步舞 ​

0. 导读：Consumer 比 Producer 复杂十倍 ​

1. subscribe vs assign：两条订阅路线 ​

1.1 subscribe（动态分配） ​

1.2 assign（手动指定） ​

1.3 对比表 ​

2. poll 循环：消费者的「心跳」 ​

2.1 最小骨架 ​

2.2 poll() 在内部干了什么 ​

2.3 timeout 的语义 ​

2.4 max.poll.records 的取舍 ​

3. Offset 提交：4 种姿势 + 4 种坑 ​

3.1 自动提交（默认） ​

3.2 同步提交 ​

3.3 异步提交 ​

3.4 批次提交（最实用） ​

3.5 手动按位点提交（Exactly-Once 关键） ​

3.6 「消息丢失」vs「重复消费」根因表 ​

4. __consumer_offsets：Offset 的「家」 ​

4.1 它是什么 ​

4.2 Key 编码 ​

4.3 怎么查看 ​

4.4 删除 / 过期 ​

5. auto.offset.reset：找不到 offset 时怎么办 ​

6. 心跳与超时四参数 ​

6.1 各自作用 ​

6.2 关系图 ​

6.3 经典关系 ​

6.4 常见踩坑 ​

6.4.1 max.poll.interval.ms 引发 Rebalance 风暴 ​

6.4.2 短暂网络抖动触发 Rebalance ​

6.4.3 滚动重启触发 N 次 Rebalance ​

7. Consumer 与 Coordinator 的 6 步协议 ​

7.1 6 步详解 ​

7.2 时序图 ​

7.3 4 种分区分配策略（assignor） ​

8. 高级 API：seek / pause / resume ​

8.1 seek：跳到任意 offset ​

8.2 pause / resume：暂停某些分区 ​

8.2.1 处理慢任务时避免 Rebalance ​

8.2.2 背压控制 ​

8.3 commit 指定位点 ​

9. 与其他 MQ 对比 ​

10. 调优清单 ​

11. 实战脚本一览 ​

12. 小结 ​

13. 面试高频题（6 题） ​

Q1：自动提交 vs 手动提交，怎么选？分别会怎么丢/重消费？ ​

Q2：max.poll.interval.ms 和 session.timeout.ms 的区别？踩坑？ ​

Q3：Consumer 与 Coordinator 的 6 步协议是什么？Rebalance 怎么触发？ ​

Q4：__consumer_offsets 是什么？为什么用 compact 策略？ ​

Q5：CooperativeSticky 和 Sticky 的区别？为什么是 2025 年默认推荐？ ​

Q6：怎么实现「不重不漏」消费？ ​

🎬 可视化演示 ​

💻 示例代码 ​

第 5 章 Consumer 深入：把 poll() 拆成消费组协议的六步舞

0. 导读：Consumer 比 Producer 复杂十倍

1. subscribe vs assign：两条订阅路线

1.1 subscribe（动态分配）

1.2 assign（手动指定）

1.3 对比表

2. poll 循环：消费者的「心跳」

2.1 最小骨架

2.2 poll() 在内部干了什么

2.3 timeout 的语义

2.4 max.poll.records 的取舍

3. Offset 提交：4 种姿势 + 4 种坑

3.1 自动提交（默认）

3.2 同步提交

3.3 异步提交

3.4 批次提交（最实用）

3.5 手动按位点提交（Exactly-Once 关键）

3.6 「消息丢失」vs「重复消费」根因表

4. `__consumer_offsets`：Offset 的「家」

4.1 它是什么

4.2 Key 编码

4.3 怎么查看

4.4 删除 / 过期

5. auto.offset.reset：找不到 offset 时怎么办

6. 心跳与超时四参数

6.1 各自作用

6.2 关系图

6.3 经典关系

6.4 常见踩坑

6.4.1 `max.poll.interval.ms` 引发 Rebalance 风暴

6.4.2 短暂网络抖动触发 Rebalance

6.4.3 滚动重启触发 N 次 Rebalance

7. Consumer 与 Coordinator 的 6 步协议

7.1 6 步详解

7.2 时序图

7.3 4 种分区分配策略（assignor）

8. 高级 API：seek / pause / resume

8.1 seek：跳到任意 offset

8.2 pause / resume：暂停某些分区

8.2.1 处理慢任务时避免 Rebalance

8.2.2 背压控制

8.3 commit 指定位点

9. 与其他 MQ 对比

10. 调优清单

11. 实战脚本一览

12. 小结

13. 面试高频题（6 题）

Q1：自动提交 vs 手动提交，怎么选？分别会怎么丢/重消费？

Q2：`max.poll.interval.ms` 和 `session.timeout.ms` 的区别？踩坑？

Q3：Consumer 与 Coordinator 的 6 步协议是什么？Rebalance 怎么触发？

Q4：`__consumer_offsets` 是什么？为什么用 compact 策略？

Q5：CooperativeSticky 和 Sticky 的区别？为什么是 2025 年默认推荐？

Q6：怎么实现「不重不漏」消费？

🎬 可视化演示

💻 示例代码