4、消费者客户端：Consumer API、消费组与重平衡、位移提交、反序列化器

好，咱们今天聊聊 Kafka 消费者。说实话，生产环境中大部分问题都出在消费者端。生产者那边只要集群稳定、网络不抖，基本不会出大乱子。但消费者？嗯，坑多得很。

我刚开始用 Kafka 那会儿，觉得消费者 API 挺简单的——不就是 poll 一下拿数据嘛。结果线上第一个版本就踩了重平衡的坑，整个消费组停了十几秒。从那以后，我再也不敢小看消费者了。

4.1 Consumer API 基础用法

Kafka 消费者最核心的操作就是订阅主题、拉取消息。说白了，就是一个循环：subscribe() 然后 poll()。

先看一段最基础的代码：

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
        StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n",
                record.offset(), record.key(), record.value());
    }
}

这段代码看着简单，但有几个细节我提醒一下：

  • poll() 的超时时间：100 毫秒是常用的值。太短了空转浪费 CPU，太长了影响消费延迟。我个人习惯设 100~200ms。
  • 循环不能停：消费者必须持续 poll，否则超过 max.poll.interval.ms（默认 5 分钟）会被判定为死亡，触发重平衡。
  • 关闭资源：生产环境记得加 finally 块调用 consumer.close()，否则连接泄漏。
小技巧：如果你用 Spring Kafka，@KafkaListener 注解帮你封装了 poll 循环和位移提交。但底层原理还是这套 API，出了问题排查时，你得知道它背后怎么跑的。

4.2 消费组与重平衡

消费组是 Kafka 实现水平扩展的核心机制。你想想看，一个主题有 10 个分区，单机消费太慢怎么办？加机器。把消费者放到同一个组里，Kafka 会自动分配分区。

我画了一张图，帮你理解消费组的分区分配逻辑：

消费组与分区分配示意图 主题：my-topic（10 个分区） P0 P1 P2 P3 P4 P5 P6 P7 P8 P9 消费组：my-group 消费者 A 分配：P0,P1,P2,P3 消费者 B 分配：P4,P5,P6 消费者 C 分配：P7,P8,P9 关键规则： • 一个分区只能被同一个消费组内的一个消费者消费 • 消费者数量 > 分区数时，多余的消费者会空闲（浪费资源）

重平衡（Rebalance）是什么？说白了就是消费组成员发生变化时，Kafka 重新分配分区的过程。触发条件有三个：

  1. 消费者加入或离开：比如新启动一个消费者，或者某个消费者挂了。
  2. 主题分区数变化：你给主题增加了分区。
  3. 消费者超时：超过 session.timeout.ms（默认 45 秒）没发心跳。
注意：重平衡期间，整个消费组会停止消费。如果分区数多、消费者多，这个过程可能持续几十秒。我曾经在线上遇到过因为 GC 停顿导致消费者心跳超时，触发重平衡，然后重平衡又加剧 GC……恶性循环，整个消费组停了近两分钟。

怎么减少重平衡的影响？我建议你关注这几个参数：

参数 默认值 说明
session.timeout.ms 45000 心跳超时时间。网络不稳定时可以适当调大
heartbeat.interval.ms 3000 心跳间隔。建议设为 session.timeout.ms 的 1/3
max.poll.interval.ms 300000 两次 poll 的最大间隔。处理慢的业务可以调大
partition.assignment.strategy RangeAssignor 分区分配策略。推荐用 StickyAssignor，减少重平衡时的分区变动

4.3 位移提交

位移提交，就是消费者告诉 Kafka：「我已经处理到 offset=X 了，下次从这个位置继续消费。」

这里有个关键选择：自动提交还是手动提交？

自动提交很简单，设置 enable.auto.commit=true（默认就是 true），然后配个 auto.commit.interval.ms（默认 5 秒）。Kafka 会在后台定时提交当前 poll 到的最大 offset。

但自动提交有个坑：可能丢数据或重复消费。举个例子，你 poll 了一批消息，处理到一半程序挂了。自动提交还没触发，重启后从上次提交的 offset 开始消费，那已经处理完的消息会重复处理。反过来，如果自动提交先触发了，但消息还没处理完，程序挂了，那这些消息就丢了。

所以生产环境，我强烈建议用手动提交：

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理业务逻辑
        process(record);
    }
    // 同步提交，确保处理完才提交
    consumer.commitSync();
}

手动提交有两种方式：

  • commitSync()：同步阻塞，提交成功才返回。可靠性高，但吞吐量受影响。
  • commitAsync()：异步提交，不阻塞。吞吐量高，但可能提交失败。

我个人习惯的做法是：正常流程用 commitAsync()，程序退出前或发生异常时用 commitSync() 兜底。这样既保证了吞吐量，又不会丢位移。

进阶技巧：如果你需要更细粒度的控制，可以用 commitSync(Map<TopicPartition, OffsetAndMetadata>) 提交指定分区的指定 offset。比如你处理完一批数据后，想精确控制每个分区的提交位置。

4.4 反序列化器

生产者把对象序列化成字节数组发到 Kafka，消费者就得反序列化回来。Kafka 内置了几种常用的反序列化器：

反序列化器 说明 适用场景
StringDeserializer 字节数组转 String 文本数据、JSON 字符串
IntegerDeserializer 字节数组转 Integer 整数类型的 key 或 value
LongDeserializer 字节数组转 Long 时间戳、ID 等长整型
ByteArrayDeserializer 直接返回原始字节数组 自定义反序列化、二进制数据

但实际项目中，我们很少直接用这些基础类型。更多时候，消息是 JSON、Avro、Protobuf 格式。这时候需要自定义反序列化器。

举个例子，假设消息是 JSON 格式的用户信息：

public class UserDeserializer implements Deserializer<User> {
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public User deserialize(String topic, byte[] data) {
        if (data == null) return null;
        try {
            return objectMapper.readValue(data, User.class);
        } catch (Exception e) {
            throw new SerializationException("反序列化失败", e);
        }
    }

    @Override
    public void close() {
        // 清理资源
    }
}

然后在配置里指定：

props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        UserDeserializer.class.getName());
避坑指南：我曾经遇到过一个线上问题——反序列化器里抛了异常，但没被捕获，导致消费者线程直接挂掉。后来我养成了习惯：反序列化器里一定要 try-catch，把异常转成可处理的业务异常，而不是让线程崩溃。另外，反序列化失败的消息可以发到死信队列（DLQ），方便后续排查。

如果你用 Avro 或 Protobuf，建议配合 Schema Registry 使用。这样生产者和消费者之间通过 schema 版本协商，避免了字段不兼容的问题。嗯，这个后面讲序列化的时候会细说。

4.5 消费者端的最佳实践

最后，我总结几条消费者端的实战经验：

  • 消费逻辑要轻量：不要在 poll 循环里做耗时操作，比如调用外部 API、写数据库。否则容易触发 max.poll.interval.ms 超时。建议用线程池异步处理。
  • 优雅关闭：用 Runtime.getRuntime().addShutdownHook() 捕获 JVM 退出信号，先调用 consumer.wakeup() 中断 poll，再 consumer.close()。
  • 监控消费延迟：通过 kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --describe 查看 LAG（积压量）。LAG 持续增长说明消费能力跟不上。
  • 幂等消费：因为可能重复消费，业务逻辑最好设计成幂等的。比如数据库操作用 UPSERT 而不是 INSERT。

消费者这块内容不少，但核心就三个词：组、位移、反序列化。把这三点搞透了，大部分消费端的问题你都能自己排查。

← 上一章 📖 返回目录 下一章 →