confluent-kafka-go消费者批量提交及重平衡处理方案问询

我平时用confluent-kafka-go做生产环境消费时，绝对不会忽略重平衡的处理——这是保证消息不重复、不丢失的核心环节，批量提交+重平衡处理是生产环境的标配方案，分享下我常用的实现思路：

• 本地维护未提交偏移量集合
  完全关闭客户端自动提交，手动控制偏移量：消费到消息并确保业务逻辑执行成功后，把对应的TopicPartition偏移量（注意要+1，因为提交的是下一条要消费的位置）存在线程安全的本地结构里，比如sync.Map或者带互斥锁的map。

• 批量提交的双触发条件
  同时满足任一条件就触发提交，兼顾性能和可靠性：

  1. 累计已处理未提交的消息数达到设定阈值（比如100条）
  2. 距离上次提交的时间超过设定间隔（比如5秒）

• 重平衡事件的关键处理
  在订阅主题时注册的回调函数里处理Assigned和Revoked事件：

  • 收到Revoked事件时：立即提交本地所有已处理的未提交偏移量——此时消费者即将放弃分区所有权，必须把已处理的进度同步到Kafka，避免后续其他消费者重复消费已处理的消息。
  • 收到Assigned事件时：一般直接用客户端默认的分区分配逻辑即可，除非有特殊业务需要手动重置偏移量。

核心代码示例 #

package main

import (
	"sync"
	"time"

	"github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
	consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": "localhost:9092",
		"group.id":          "biz-consumer-group",
		"auto.offset.reset": "earliest",
		"enable.auto.commit": false, // 强制关闭自动提交
	})
	if err != nil {
		panic(err)
	}
	defer consumer.Close()

	// 订阅主题并注册重平衡回调
	err = consumer.SubscribeTopics([]string{"biz-topic"}, func(c *kafka.Consumer, event kafka.Event) error {
		switch e := event.(type) {
		case kafka.AssignedPartitions:
			c.Assign(e.Partitions)
		case kafka.RevokedPartitions:
			// 分区回收前必须提交已处理偏移量
			commitPendingOffsets(c, &pendingOffsets)
			c.Unassign()
		}
		return nil
	})
	if err != nil {
		panic(err)
	}

	var wg sync.WaitGroup
	wg.Add(1)

	var pendingOffsets sync.Map // 存储未提交的偏移量
	commitTicker := time.NewTicker(5 * time.Second)
	defer commitTicker.Stop()

	go func() {
		defer wg.Done()
		for {
			select {
			case <-commitTicker.C:
				// 定时触发提交
				commitPendingOffsets(consumer, &pendingOffsets)
			default:
				msg, err := consumer.ReadMessage(100 * time.Millisecond)
				if err != nil {
					continue
				}

				// 先确保业务逻辑执行成功
				if handleBizLogic(msg) {
					// 记录下一条要消费的偏移量
					tp := kafka.TopicPartition{
						Topic:     msg.TopicPartition.Topic,
						Partition: msg.TopicPartition.Partition,
						Offset:    msg.TopicPartition.Offset + 1,
					}
					pendingOffsets.Store(genTpKey(tp), tp)

					// 达到数量阈值触发提交
					if countPending(&pendingOffsets) >= 100 {
						commitPendingOffsets(consumer, &pendingOffsets)
					}
				}
			}
		}
	}()

	wg.Wait()
}

// genTpKey 生成TopicPartition的唯一标识键
func genTpKey(tp kafka.TopicPartition) string {
	return *tp.Topic + "-" + string(rune(tp.Partition))
}

// countPending 统计未提交偏移量的数量
func countPending(m *sync.Map) int {
	cnt := 0
	m.Range(func(_, _ interface{}) bool {
		cnt++
		return true
	})
	return cnt
}

// commitPendingOffsets 提交偏移量并清空本地缓存
func commitPendingOffsets(c *kafka.Consumer, m *sync.Map) {
	var offsets []kafka.TopicPartition
	m.Range(func(_, val interface{}) bool {
		offsets = append(offsets, val.(kafka.TopicPartition))
		return true
	})
	if len(offsets) == 0 {
		return
	}

	// 同步提交，确保提交结果可感知
	_, err := c.CommitOffsets(offsets)
	if err == nil {
		// 提交成功后清空本地存储
		m.Range(func(key, _ interface{}) bool {
			m.Delete(key)
			return true
		})
	} else {
		// 提交失败可记录日志，等待下次提交重试
		// log.Printf("Commit offsets failed: %v", err)
	}
}

// handleBizLogic 模拟业务逻辑处理
func handleBizLogic(msg *kafka.Message) bool {
	// 替换为实际业务逻辑，返回true表示处理成功
	return true
}

额外注意事项 #

• 业务逻辑必须保证幂等性：哪怕提交偏移量失败，后续重新消费时业务要能处理重复消息，避免数据异常。
• 重平衡回调函数要保证线程安全：confluent-kafka-go的回调是在独立goroutine中执行的，操作共享资源时要注意同步。

内容的提问来源于stack exchange，提问作者Yura