消费者与消费者组

中间件

Kafka

消费者与消费者组

消费者组与分区分配、多消费者组独立消费、重平衡（Rebalance）机制与心跳，以及 ReaderConfig 消费者配置完整对照。

消费者（Consumer）

在 kafka-go 中，使用 Reader 来订阅 Topic 并接收消息。当生产者写入速度超过单个消费者的处理速度时，需要通过消费者组进行水平扩展。

// 创建一个消费者
r := kafka.NewReader(kafka.ReaderConfig{
    Brokers: []string{"localhost:9092"},
    Topic:   "order-events",
    GroupID: "order-processors", // 消费者组 ID
})
defer r.Close()

for {
    msg, err := r.ReadMessage(context.Background())
    if err != nil {
        break
    }
    fmt.Printf("收到: %s\n", string(msg.Value))
}

消费者组与分区分配

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

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

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

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

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

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

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

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

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

多消费者组

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

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

在 kafka-go 中，只需为不同应用使用不同的 GroupID 即可实现：

// 应用 A：订单处理服务
orderReader := kafka.NewReader(kafka.ReaderConfig{
    Brokers: []string{"localhost:9092"},
    Topic:   "order-events",
    GroupID: "order-service", // 独立的消费者组
})

// 应用 B：审计日志服务
auditReader := kafka.NewReader(kafka.ReaderConfig{
    Brokers: []string{"localhost:9092"},
    Topic:   "order-events",
    GroupID: "audit-service", // 另一个独立的消费者组，收到同样的全量消息
})

总结：

如果你的应用需要读取全量消息 → 为该应用创建一个独立的消费者组（使用不同的 GroupID）

如果该应用消费能力不足 → 在该消费者组内启动多个 Reader 实例（注意总数不要超过分区数）

消费者组和分区重平衡

消费者组是什么

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

两种消费模式：

模式	说明
点对点（P2P）	一个消费者组消费一个 Topic，组内各消费者竞争消费不同分区
发布-订阅（Pub-Sub）	多个消费者组各自独立消费同一 Topic 的全量消息

消费者重平衡（Rebalance）

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

重平衡是一把双刃剑：

✅ 带来高可用性（节点挂掉后自动迁移分区）和伸缩性（自由增减消费者）
❌ 重平衡期间，整个消费者组停止消费（类似于 JVM 的 Stop The World），过程较慢
❌ 分区被重新分配时，消费者当前的读取状态会丢失，可能需要刷新缓存来恢复状态

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

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

心跳与重平衡触发机制

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

重平衡的触发条件：

消费者停止发送心跳，会话超时（session.timeout.ms），组协调者判定该消费者已死亡
消费者宕机，组协调者等待几秒确认死亡后触发重平衡（在此期间该消费者不处理任何消息）
消费者主动离开群组（正常关闭时通知协调者），触发重平衡以尽量降低处理停顿

重平衡过程中的 bug 至今社区仍在持续改进。

消费者配置详解

kafka-go 的 ReaderConfig 封装了 Kafka 消费者的核心配置。以下将 Kafka Broker 端的消费者参数与 kafka-go 中对应的字段进行对照说明：

fetch.min.bytes → MinBytes

消费者从 Broker 获取数据的最小字节数。如果可用数据不足该值，Broker 会等待直到有足够数据才返回。

r := kafka.NewReader(kafka.ReaderConfig{
    MinBytes: 10e3,  // 最少拉取 10KB
    MaxBytes: 10e6,  // 最多拉取 10MB
})

fetch.max.wait.ms → MaxWait

Broker 等待数据的最大时间，默认无上限。如果数据量达不到 MinBytes，Broker 最多等待该时长后就返回所有可用数据。

r := kafka.NewReader(kafka.ReaderConfig{
    MaxWait: 500 * time.Millisecond, // 最多等 500ms
})

max.partition.fetch.bytes → MaxBytes

Broker 从每个分区返回给消费者的最大字节数。MaxBytes 限制了单次 ReadMessage 能拉取的总数据量。

重要：MaxBytes 必须大于 Broker 端 message.max.bytes 的值，否则可能无法读取某些大消息。

session.timeout.ms → SessionTimeout

消费者被认为死亡前可与 Broker 断开连接的最长时间。

r := kafka.NewReader(kafka.ReaderConfig{
    SessionTimeout: 10 * time.Second,  // 10 秒无心跳则认为消费者已挂
    HeartbeatInterval: 3 * time.Second, // 每 3 秒发一次心跳（应 < SessionTimeout）
})

auto.offset.reset → StartOffset

在消费者没有 Offset 或 Offset 无效时的处理策略：

// 从最早的消息开始（首次消费或 Offset 过期后）
r := kafka.NewReader(kafka.ReaderConfig{
    StartOffset: kafka.FirstOffset,
})

// 仅从最新的消息开始（默认行为）
r := kafka.NewReader(kafka.ReaderConfig{
    StartOffset: kafka.LastOffset,
})

enable.auto.commit → 隐式行为

kafka-go 中没有直接对应的布尔开关，而是通过 API 选择来控制：

方式	说明
`ReadMessage(ctx)`	自动提交。每次成功返回后自动将 Offset 提交到 Broker
`FetchMessage(ctx)` + `CommitMessages(ctx, msg)`	手动提交。完全由你控制提交时机

partition.assignment.strategy → GroupBalancers

分区分配策略，kafka-go 支持同时配置多个策略（按优先级排列）：

r := kafka.NewReader(kafka.ReaderConfig{
    GroupBalancers: []kafka.GroupBalancer{
        kafka.RoundRobinGroupBalancer{}, // 优先使用轮询
        kafka.RangeGroupBalancer{},      // 备选范围分配
    },
})

Balancer	说明
`RangeGroupBalancer`	按 Topic 分区范围分配（默认）
`RoundRobinGroupBalancer`	轮询分配所有分区

max.poll.records → 无直接对应

kafka-go 的 ReadMessage 每次只返回一条消息。若需批量拉取，需要在for循环中自行控制拉取条数，或使用底层的 ReadBatch。

client.id → 无直接对应

kafka-go 目前不直接暴露 client.id 设置，Broker 端日志中以连接 IP 标识客户端。

ReaderConfig 完整配置速查

r := kafka.NewReader(kafka.ReaderConfig{
    // ─── 连接 ───
    Brokers:     []string{"localhost:9092", "localhost:9093"},
    Topic:       "my-topic",
    GroupID:     "my-group",         // 消费者组 ID
    Partition:   0,                  // 指定分区（与 GroupID 互斥）

    // ─── 拉取控制 ───
    MinBytes:        1,             // 最少拉取字节数
    MaxBytes:        10e6,          // 最多拉取字节数 (10MB)
    MaxWait:         5 * time.Second, // 最大等待时间
    ReadBatchTimeout: 3 * time.Second, // 内部批量读取超时

    // ─── Offset 管理 ───
    StartOffset:     kafka.FirstOffset, // 起始位置
    CommitInterval:  time.Second,       // 自动提交间隔（0 = 每次 ReadMessage 都提交）

    // ─── 心跳与会话 ───
    SessionTimeout:    30 * time.Second,
    HeartbeatInterval: 3 * time.Second,
    RebalanceTimeout:  30 * time.Second,
    RetentionTime:     24 * time.Hour, // 消费者组 Offset 保留时长

    // ─── 其他 ───
    MaxAttempts:     3,        // 网络错误重试次数
    ReadLagInterval: 3 * time.Second, // 消费延迟检测间隔
    GroupBalancers:  []kafka.GroupBalancer{
        kafka.RoundRobinGroupBalancer{},
    },
    Logger: kafka.LoggerFunc(log.Printf), // 日志输出
    ErrorLogger: kafka.LoggerFunc(log.Printf),
})

生产者偏移量与提交