消费调优:Fetch 参数,并发模型与背压

一个慢消费的现场

订单处理服务:每条消息要调两次下游 RPC,各 50ms,单条处理耗时 100ms.Topic 6 分区,3 个消费者,按第 04 篇的"每分区一个 goroutine"消费.

处理能力算一下:

单分区串行处理: 1000ms / 100ms = 10 条/s
单消费者 2 分区: 20 条/s
整个组 3 消费者: 60 条/s

上游生产速率:   200 条/s

缺口 140 条/s,lag 单调上涨.不处理的结局是第 07 篇讲过的:lag 涨到 retention 之外,老消息被删,数据丢失.

这个现场里可调的只有三个维度:

动手之前先把瓶颈分段,拉住"拉"和"处理"两个环节:

lag 持续增长, 先分清瓶颈在哪一段:

  records-consumed-rate 上不去, CPU 不饱和
      → 拉得慢: 网络 / fetch 参数 / broker 端(第 07 篇)

  consumed rate 正常, lag 不降, 内存上涨
      → 处理不完: 并发不够 / 单条太慢 / 下游是瓶颈

Fetch 参数:一次拉多少

第 04 篇讲过 Pull 模型与 long poll,这里把一轮 FetchRequest 的参数补全:

消费者 FetchRequest                          Broker
   │  FetchMinBytes=1, FetchMaxWait=500ms      │
   ├────────────────────────────────────────────►│
   │                                             │ 数据不足 minBytes 时挂起等待
   │                                             │ (最长等 MaxWait)
   │◄────────────────────────────────────────────┤ 返回 ≤ FetchMaxBytes 的数据
   │  (单个分区单次 ≤ FetchMaxPartitionBytes)     │

参数对照(Java 名 / franz-go 选项 / 作用 / Java 默认值):

四个调优点:

1. FetchMinBytes 调大(如 1MB):Broker 攒够再返回,少而大的响应,省往返.吞吐型消费的常用手段;延迟敏感场景保持小值
2. FetchMaxPartitionBytes 是双刃剑:默认 1MB,大消息或大积压时单个分区要多次往返才能拉完;调大加快拉取,但单次 Poll 的内存占用也上涨
3. FetchMaxBytes 是内存闸门:一次拿回的总量上限,处理慢的消费者别调太大
4. MaxPollRecords 与处理速度有明确算式:

MaxPollRecords × 单条处理时间 < max.poll.interval.ms

例: 单条 100ms, max.poll.interval 默认 5min
  → 单批上限约 3000 条
  → MaxPollRecords 设 5000 且批批都要处理完, 必然超时被踢(第 04/11 篇)

这就是"fetch 参数与处理速度匹配"的全部含义:一次拉回的活,必须在 poll 超时前干完.

Worker Pool:并发模型落地

第 04 篇给过三种模式的骨架,这里给出可落地的实现,重点是 offset 提交的正确处理.

方案一:每分区一个 goroutine,分区内保序

顺序敏感场景的标准解.每个分区一个 goroutine,提升单分区吞吐靠"批量化处理":

fetches.EachPartition(func(p kgo.FetchTopicPartition) {
    wg.Add(1)
    go func(topic string, partition int32, records []*kgo.Record) {
        defer wg.Done()
        // 关键优化: 慢的是 RPC, 不是计算 → 攒批调用
        // 50ms 调 1 条 vs 50ms 调 100 条, 吞吐差两个数量级
        processBatch(records)
    }(p.Topic, p.Partition, p.Records)
})
wg.Wait()
client.CommitUncommittedOffsets(ctx)

并发度 = 分区数,分区内串行(保序),offset 提交简单:整批处理完统一提交.开头现场里的 100ms 单条 RPC,批量化后可能变成 50ms 处理 100 条.

方案二:共享 worker pool,吞吐优先

任何分区的消息交给任意 worker,并发度不受分区数限制,但分区内顺序丢失(第 02 篇:有序性只在分区内).实现的关键是 offset 提交不能"处理完一条提交一条":

Partition 0 各 offset 的处理状态:

offset 100 ✓   101 ✓   102 ✗(处理中)   103 ✓
                      ▲
        连续完成的最高水位是 101
        → 只能提交到 102, 不能提交到 104

若 103 被提前提交而 102 之后崩溃:
  → 重启后从 103 之后继续, 102 被跳过 → 丢消息

正确做法是维护"连续完成水位":

// 每个分区维护: 已完成集合 + 连续水位
type tracker struct {
    mu       sync.Mutex
    done     map[int64]bool  // 已完成但尚未连成片
    highmark int64           // 连续完成的最高 offset
}

func (t *tracker) mark(offset int64) int64 {
    t.mu.Lock()
    defer t.mu.Unlock()
    t.done[offset] = true
    for t.done[t.highmark+1] {   // 从水位开始, 连续推进
        delete(t.done, t.highmark+1)
        t.highmark++
    }
    return t.highmark
}

提交时只提交各分区的 highmark + 1.复杂度不低,换来的是吞吐上限只受 worker 数约束.

方案三:有界队列 + 阻塞投递(最简背压)

records := make(chan *kgo.Record, 1000)   // 有界队列

// worker 侧
for i := 0; i < numWorkers; i++ {
    go func() {
        for r := range records {
            process(r)
        }
    }()
}

// poll 侧
for {
    fetches := client.PollFetches(ctx)
    fetches.EachRecord(func(r *kgo.Record) {
        records <- r   // 队列满, 这里阻塞 → 拉取速度自动降到处理速度
    })
    // 注意: 此时不能立即提交 offset(队列里还有在途消息)
}

代价是 offset 提交又回到"有在途消息未完成"的问题,需要配合完成计数或水位追踪.前两个方案的处理能力不够时才上这一层.

背压处理:数据堆在哪

三种做法的对比:

franz-go 提供了显式的暂停接口,适合"本地队列 + 水位"的精细控制:

// 本地积压超过高水位: 暂停拉取
if localQueue.Len() > highWatermark {
    client.PauseFetchTopics("order-events")
}

// 消化到低水位: 恢复
if localQueue.Len() < lowWatermark {
    client.ResumeFetchTopics("order-events")
}

需要更细粒度时还有分区级的 PauseFetchPartitions / ResumeFetchPartitions.

慢消费治理:按顺序排查

1. 量化: lag 增速 = 生产速率 - 消费速率; 清空时间 = lag / 净消费速率
2. 分段: 瓶颈在"拉"还是"处理"(本文开头的判断树)
3. 处理侧按性价比选手段(见下表)
4. 检查连锁反应: max.poll.interval 是否被突破(触发 rebalance)

处理侧的手段,按投入产出排序:

两个容易忽略的点:

• 加消费者有上限:消费者数超过分区数,多出来的实例拿不到分区(第 04 篇),再加只是浪费资源
• 慢消费与 rebalance 是连锁的:单批处理时间超过 max.poll.interval.ms,分区被夺走,进入第 11 篇的排查清单.止血手段之一就是把 MaxPollRecords 调小,让单批更快回到 Poll

动手验证

# 1. 制造消费压力: perf-test 持续生产(第 12 篇的命令)
kafka-producer-perf-test.sh --topic order-events \
  --num-records 500000 --record-size 512 --throughput 2000 \
  --producer-props bootstrap.servers=localhost:19092

# 2. 观察 lag(第 07 篇的命令)
kafka-consumer-groups.sh --describe --group order-processor \
  --bootstrap-server localhost:19092

代码侧对照实验:

1. 处理函数 sleep 100ms, 分别用 1 / 4 / 16 个 worker, 观察 lag 曲线
2. MaxPollRecords 设成让单批必然超时, 观察分区被夺走与 rebalance(接第 11 篇)
3. 有界队列调小(如 100), 观察消费速率自动降到处理速度、内存保持稳定
4. 把逐条 RPC 改成攒批 RPC, 对比同样并发下 lag 清空时间的数量级变化

生产注意事项

下一篇是本阶段最后一篇:事务与 Exactly-Once.消费端"不丢不重"的完整答案在那里,跨分区原子写入、事务型生产者与消费者,以及 Kafka 事务的能力边界.