Skip to content
中间件
Kafka
偏移量与提交

偏移量与提交

__consumer_offsets 内部 Topic、Offset 提交不当的后果（重复消费/消息丢失），以及 kafka-go 中自动提交与手动提交的实现方式。

特殊 Topic：__consumer_offsets

Kafka 通过一个名为 __consumer_offsets 的内部 Topic 来记录每个消费者组在每个分区上的消费进度（Offset）。

__consumer_offsets 的主要作用是：当消费者组发生重平衡后，每个消费者可以从上次提交的 Offset 处继续消费，避免消息丢失或重复。

在 kafka-go 中，Reader 自动管理 Offset——当使用 ReadMessage 时，每次成功读取后会自动提交当前 Offset。你也可以通过 FetchMessage + CommitMessages 获得完全的手动控制。

Offset 提交不当的后果

提交的 Offset 小于实际消费到的位置 → 消息被重复消费

img

提交的 Offset 大于实际消费到的位置 → 中间的消息丢失

img

Offset 提交方式（kafka-go 实现）

kafka-go 提供了两种消费模式，对应不同的 Offset 提交策略：

1. 自动提交 — ReadMessage

最简单的方式。使用 ReadMessage 时，每次成功读取消息后自动提交 Offset。提交频率可通过 CommitInterval 控制：

// 每次 ReadMessage 成功后自动提交（CommitInterval = 0）
r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:  []string{"localhost:9092"},
    Topic:    "my-topic",
    GroupID:  "my-group",
    CommitInterval: 0, // 0 表示每次读取后立即提交
})

for {
    msg, err := r.ReadMessage(ctx) // 读取 + 自动提交
    if err != nil {
        break
    }
    fmt.Printf("收到: %s\n", string(msg.Value))
    // 无需手动提交
}

也可以设置 CommitInterval 来批量提交以提升性能：

r := kafka.NewReader(kafka.ReaderConfig{
    CommitInterval: time.Second, // 每秒批量提交一次，减少网络开销
})

优点：简单省心，适合大多数场景 缺点：如果在自动提交间隔内发生重平衡，已读取但未提交的消息会被重复消费

2. 手动提交（精确控制）— FetchMessage + CommitMessages

将 Offset 的提交时机完全交给应用层控制——只有当消息业务处理成功后，才提交：

ctx := context.Background()
for {
    // FetchMessage 拉取消息，不提交 Offset
    msg, err := r.FetchMessage(ctx)
    if err != nil {
        break
    }

    fmt.Printf("message at offset %d: %s = %s\n",
        msg.Offset, string(msg.Key), string(msg.Value))

    // 处理业务逻辑...
    if err := processMessage(msg); err != nil {
        // 处理失败 → 不提交，消息会被重新消费
        log.Printf("处理失败: %v", err)
        continue
    }

    // 处理成功 → 确认提交（同步提交，会重试直到成功）
    if err := r.CommitMessages(ctx, msg); err != nil {
        log.Fatal("提交 Offset 失败: ", err)
    }
}

3. 组合策略：先处理，再提交通常做法

平时使用 FetchMessage 处理，关闭时使用 ReadMessage 的自动提交兜底，确保最后一批消息不丢失：

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

go func() {
    // 正常运行时：手动控制提交
    for {
        msg, err := r.FetchMessage(ctx)
        if err != nil {
            return
        }
        if err := processMessage(msg); err == nil {
            r.CommitMessages(ctx, msg)
        }
        // 失败的消息不提交，下次重平衡后重新消费
    }
}()

// 等待退出信号
<-sig
cancel()
r.Close() // 优雅关闭，确保未提交的消息不会丢失

提交规则：提交某个 Offset 时，该分区所有更小的 Offset 也会被一并标记为已消费——与 TCP 的累积确认机制类似。

Java vs kafka-go Offset 提交对照

Java Kafka Consumer kafka-go 说明
poll() 拉取记录 ReadMessage() 或 FetchMessage() 都是拉取单条
enable.auto.commit=true 使用 ReadMessage() 自动提交
enable.auto.commit=false 使用 FetchMessage() + CommitMessages() 手动提交
commitSync() CommitMessages() 同步提交（会重试）
commitAsync() 无直接对应 kafka-go 不支持异步提交，需要自行开 goroutine
commitSync(offsets) 提交特定 Offset CommitMessages(ctx, msg) 按消息提交，底层等价
auto.commit.interval.ms CommitInterval 批量提交的时间间隔
消费者与消费者组MQ 选型与架构