消费者与消费者组

消费者组与分区分配、多消费者组独立消费、重平衡（Rebalance）机制与协作式协议/静态成员、消费积压监控，以及 franz-go 消费者配置完整对照（kafka-go 存量注记保留）。

消费者（Consumer）

franz-go 的消费者就是"带组选项的客户端"。当生产者写入速度超过单个消费者的处理速度时，通过消费者组水平扩展：

group, err := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumerGroup("order-processors"), // 消费者组 ID
    kgo.ConsumeTopics("order-events"),
    kgo.ConsumeStartOffset(kgo.NewOffset().AtEnd()), // 新组起点：显式写，别用默认（下文详述）
)
if err != nil {
    log.Fatal(err)
}
defer group.Close()

for {
    fs := group.PollFetches(ctx)
    fs.EachRecord(func(r *kgo.Record) {
        fmt.Printf("收到: %s\n", string(r.Value))
    })
}

kafka-go 存量对应 kafka.NewReader(kafka.ReaderConfig{Brokers, Topic, GroupID, ...}) + ReadMessage/FetchMessage，模型差异见 03 章速查。

消费者组与分区分配

下面通过一组图来理解消费者组的工作原理：

阶段一：1 个消费者处理全部 4 个分区的消息

消费者组 1 中只有一个消费者，它订阅了 Topic T1，接收全部四个分区的消息。压力较大。

阶段二：增加消费者分担压力

增加了一个消费者后，每个消费者各自负责两个分区，消费能力成倍提升。

阶段三：继续增加消费者达到最佳状态

四个消费者各负责一个分区，达到理想的一对一状态。

阶段四：消费者数量超过分区数

此时第 5 个消费者会被闲置。消费者的数量不应超过分区数，多出来的消费者没有任何作用。

多消费者组

Kafka 的一个重要特性是：只需写入一次消息，可以支持任意多个应用独立消费全量消息。

在这个场景中，消费组 G1 和 G2 各自独立地收到 T1 的全量消息，逻辑上属于两个不同的应用。

在代码里，只需为不同应用使用不同的组名即可实现：

orderGroup, _ := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumerGroup("order-service"), // 应用 A：订单处理服务
    kgo.ConsumeTopics("order-events"),
)

auditGroup, _ := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumerGroup("audit-service"), // 应用 B：审计日志服务，收到同样的全量消息
    kgo.ConsumeTopics("order-events"),
)

总结：

• 如果你的应用需要读取全量消息 → 为该应用创建一个独立的消费者组（使用不同的组名）
• 如果该应用消费能力不足 → 在该消费者组内启动多个消费实例（注意总数不要超过分区数）

消费者组和分区重平衡

消费者组是什么

消费者组（Consumer Group） 是由一个或多个消费者实例组成的群组，是实现可扩展性和可容错性的核心机制。组内消费者共享一个组名，共同消费一个 Topic，同组内的每个消费者各负责一部分分区，多余的消费者会闲置。

两种消费模式：

消费者重平衡（Rebalance）

从上面的消费者演变过程可以看出：最初一个消费者处理全部分区，后来新消费者加入并分摊了部分分区的消息。这种将分区所有权从一个消费者转让给其他消费者的过程称为重平衡（Rebalance）。

重平衡是一把双刃剑：

• 带来高可用性（节点挂掉后自动迁移分区）和伸缩性（自由增减消费者）
• 但重平衡期间消费会停顿，过程较慢——尤其是旧式（eager）协议：所有成员先放弃全部分区、再重新领取，整个组经历一次"Stop The World"
• 分区被重新分配时，消费者当前的读取状态会丢失，可能需要刷新缓存来恢复状态

引用《深入理解 Java 虚拟机》中对 STW 的描述：

“它在进行垃圾收集时，必须暂停其他所有工作线程，直到收集结束。Stop The World 这个名字听起来很酷，但这项工作实际上是由虚拟机在后台自动发起并完成的，在用户不可见的情况下把用户正常工作的线程全部停掉，这对很多应用来说都是难以接受的。”

协作式协议（下面进阶小节）出现后，“整组 STW"只是 eager 协议的特征，不再是所有重平衡的宿命——这也是本手册把主线放在 franz-go（默认协作式）的原因之一。

心跳与重平衡触发机制

消费者通过向组协调者（Group Coordinator，某个 Broker）发送心跳来维持自己在组内的成员资格。只要定期发送心跳，就会被认为存活并正常处理分区消息。

重平衡的触发条件：

• 消费者停止发送心跳，会话超时（SessionTimeout，franz 默认 45s），协调者判定该消费者已死亡
• 消费者宕机，协调者等待会话超时后触发重平衡（在此期间该消费者名下的分区无人消费）
• 消费者主动离开群组（正常关闭时通知协调者），触发重平衡以尽量降低处理停顿

franz-go 与 kafka-go 都用心跳后台任务维持成员资格，PollFetches 之后处理消息再慢也不会影响"存活证明”——与 Java 客户端 max.poll.interval.ms（单批处理超时直接被踢出组）的行为不同。别把这当纯优点：它同时失去了"卡死自动转移分区"的保护，卡死的消费者会一直占着分区，看门狗要自己配（消费循环带超时、Lag 告警兜底）。

重平衡过程中的 bug 至今社区仍在持续改进（KIP-848 把整个协议搬到 Broker 端就是这条线上的终点，见下文）。

消费者配置详解

franz-go 的消费者选项与 Kafka Java 参数的映射（默认值实测自 v1.22.2 源码）：

fetch.min.bytes / fetch.max.wait.ms → FetchMinBytes / FetchMaxWait

Broker 攒够 FetchMinBytes 才返回，最多等 FetchMaxWait（franz 普通消费组默认 5s；kafka-go MaxWait 默认 10s）。低流量 Topic 上这两个值直接决定消息延迟下限。

group, _ := kgo.NewClient(
    /* ...组选项... */
    kgo.FetchMinBytes(1),
    kgo.FetchMaxWait(time.Second), // 对延迟敏感就调小
)

max.partition.fetch.bytes → FetchMaxPartitionBytes / FetchMaxBytes

单分区/单请求拉取字节上限（franz 默认 1MB / 50MB）。拉不到大消息、Fetches 报错，先对照这三处限额（生产者侧链条见 04 章）。kafka-go 用户注意：它的 MaxBytes 默认只有 1MB，大消息体必须显式调大——这是存量事故高发点。

auto.offset.reset → ConsumeStartOffset / ConsumeResetOffset

franz 拆成了两个选项：新组起点（默认 AtStart()＝earliest）与位点失效时的恢复策略（默认 RewindOffset(1min)，回退重放不丢数据）。映射表见 03 章。

注意：kafka-go 的 StartOffset 零值同样等价 FirstOffset——与 Java 默认 latest 相反，新组第一次上线会从 Topic 头开始全量重放，下游扛不住的话就是一次事故。不论用哪个客户端，新组上线必须显式写起点。

enable.auto.commit → 提交三档

franz 用选项组合表达：默认自动提交（AutoCommitInterval 5s）→ AutoCommitMarks() 标记式 → DisableAutoCommit() + CommitRecords 全手动。“至多一次/至少一次"的语义推演在 06/08 章；kafka-go 对应 ReadMessage vs FetchMessage+CommitMessages。

partition.assignment.strategy → Balancers

franz 默认只有一个：CooperativeStickyBalancer()（协作式增量分配）。可换 RangeBalancer()/RoundRobinBalancer()/StickyBalancer()，或为 KIP-848 配 ServerSideBalancer()。kafka-go 只实现了 eager 协议（RangeGroupBalancer/RoundRobinGroupBalancer），没有协作式和静态成员——这是存量 kafka-go 组在滚动发布时 Lag 锯齿的根源。

max.poll.records → PollRecords(ctx, n)

franz 在每次调用时限制条数（对照 Java 的配置项更灵活：同一客户端不同场景可以不同）。

client.id → ClientID

franz 用 kgo.ClientID("order-svc")（03 章）；kafka-go 的 Reader/Writer 不暴露 client.id（低层 Conn 有），Broker 端日志只能以连接 IP 标识。

franz-go 消费侧选项速查

group, _ := kgo.NewClient(
    // ─── 连接与身份 ───
    kgo.SeedBrokers("localhost:9092", "localhost:9093"),
    kgo.ClientID("order-svc"),
    kgo.ConsumerGroup("order-processors"),
    kgo.InstanceID("order-processors-2"), // 静态成员：稳定唯一，重启不触发重平衡

    // ─── 订阅 ───
    kgo.ConsumeTopics("order-events"),
    kgo.ConsumeRegex(),                   // 订阅按正则展开（kafka-go 无此能力）

    // ─── 起点与重置 ───
    kgo.ConsumeStartOffset(kgo.NewOffset().AtEnd()),        // 新组起点（默认 AtStart=earliest，务必显式写）
    kgo.ConsumeResetOffset(kgo.RewindOffset(time.Minute)),  // 越界/丢数据恢复策略

    // ─── 拉取控制 ───
    kgo.FetchMinBytes(1),
    kgo.FetchMaxWait(5 * time.Second),          // 默认：普通消费组 5s
    kgo.FetchMaxBytes(50 << 20),                // 单请求上限（默认 50MB）
    kgo.FetchMaxPartitionBytes(1 << 20),        // 单分区上限（默认 1MB）
    kgo.MaxConcurrentFetches(2),                // 并发 fetch 数（默认不限）

    // ─── 提交 ───
    kgo.AutoCommitInterval(5 * time.Second),    // 自动提交周期（默认 5s）
    // 或 kgo.AutoCommitMarks() / kgo.DisableAutoCommit()（见上文三档）

    // ─── 事务与隔离 ───
    kgo.FetchIsolationLevel(kgo.ReadCommitted()), // 只读已提交事务消息

    // ─── 会话与重平衡 ───
    kgo.SessionTimeout(45 * time.Second),
    kgo.HeartbeatInterval(3 * time.Second),
    kgo.RebalanceTimeout(60 * time.Second),
    kgo.Balancers(kgo.CooperativeStickyBalancer()),
    // KIP-848：kgo.ServerSideBalancer()（Kafka 4.0+，建议 4.3+，见下文）
)

对照记忆：kafka-go 的 ReaderConfig 里 RetentionTime（组位点保留）、ReadLagInterval、GroupBalancers 多路回退这些字段在 franz 没有对应——前者是 Broker 端 offsets.retention.minutes 的事（06 章），后两者 franz 根本不需要（监控走 kadm、分配器没有 eager 回退问题）。

重平衡进阶：协作式协议与静态成员

实际运维中两类高频问题：

① 滚动发布风暴。 K8s 里一次 rollout 重启 5 个消费者 Pod，eager 协议下会连环触发多次全组重平衡，期间整个组完全停止消费。现代解法是 Kafka 2.4+ 的两项协议能力，franz-go 默认/原生都支持：

• 协作式重平衡（Cooperative Rebalance）：增量式，成员只放弃"要分给别人"的分区，其余分区继续消费，不再全员 STW。franz 默认 CooperativeStickyBalancer() 就是它（Java CooperativeStickyAssignor 同名能力）。同一个组内所有成员必须都用协作式实现才能生效——混跑 kafka-go（只有 eager）时协议会退回整体分配，所以升级期间按"整组换客户端"而不是"单实例换"来规划。
• 静态成员（Static Membership）：kgo.InstanceID("order-processors-2") 给实例一个跨重启稳定的身份，K8s 下取 StatefulSet Pod 名即可——同一身份的实例重启不触发重平衡，只是把它名下的分区原样还回去。代价是"挂了要等会话超时才把分区让出来”，弹性伸缩变迟钝，适合"发布频繁、故障罕见"的稳态组。ID 必须每实例唯一且稳定，全员同名等于互相顶掉。

② 新一代组协议 KIP-848（Kafka 4.0+）。 重平衡从"客户端算、Coordinator 收作业"变成"Coordinator（Broker 端）直接算"：心跳协议更轻、分配决策集中、未来特性（如 Share Groups 的逐条 ACK）都长在这边。franz-go 接入方式是加 kgo.ServerSideBalancer() 并把分配器设为 RangeBalancer() 或 StickyBalancer()。版本提示：Kafka 4.3 之前重平衡期间提交位点可能遇到 STALE_MEMBER_EPOCH（franz 文档明确建议 4.3+ 再开）。kafka-go 不会支持这一代协议——这是"存量简单链路、新特性找 franz"分界线最硬的一刀。

消费积压怎么监控

三条路，按精度和成本选：

// ① 平台级自采：kadm 直接算组 Lag（等价 --describe 的 LAG 列汇总），配 Prometheus 推送
adm := kadm.NewClient(group)
lags, err := adm.Lag(ctx, "order-processors")
for _, dl := range lags.Sorted() {
    if dl.State != "Stable" {
        log.Printf("group=%s 处于 %s，非 Stable", dl.Group, dl.State) // 重平衡中本身也值得上报
    }
    log.Printf("group=%s total_lag=%d", dl.Group, dl.Total())
}

// ② 进程内自检：franz 没有 kafka-go 的 Stats() 习惯用法，更常见的是埋自己代码的计数器
//    （每秒 handle 条数、CommitRecords 错误数）；Fetches 里有每个分区的 lag：
fs.EachRecord(func(r *kgo.Record) { /* r 上没有 lag；用 fs 的 Fetch 级信息或定期 ① 兜底 */ })

// ③ 组外视角：Broker 侧周期性 --describe 消费者组
//    .\bin\windows\kafka-consumer-groups.bat --bootstrap-server ... --describe --group ...
//    适合导出器/巡检脚本，不受消费者进程死活影响。

kafka-go 存量注意两点：组模式 Reader 的 Lag/ReadLag/Stats().Lag 返回 -1（03 章限制表），别照搬监控代码；franz 侧无此限制（走 ①）。

注意：Lag 报警阈值定"条数"不如定"时间"：LAG / 历史消费速率 = 预计追平时长，追平时长超过业务容忍窗口才报警——一个瞬时尖峰冲上 10 万条但 2 分钟能消化完的 Topic，按条数阈值会天天误报。

优雅关闭

franz 的消费端退出三步（kafka-go 的 Reader.Close() 同理，见 03 章）：

sig := make(chan os.Signal, 1)
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
<-sig

cancel()    // 1. 让 Poll 随 ctx 退出（Err0 检出 context.Canceled 后收尾在途批次）
group.Close() // 2. 主动离组：不等会话超时，组内其他成员立刻可以接盘重平衡

手动提交模式下，退出前把已处理完的记录 CommitRecords 掉，把重放的量压到最小；K8s 里给足 terminationGracePeriodSeconds。