作为一名普通的程序开发者，我在日常工作中经常使用 Kafka 进行消息队列的处理。在一次项目中，我遇到了一个比较复杂的 Kafka 消费者组消费异常的问题，虽然看似简单，但排查过程却让我对 Kafka 的消费机制有了更深入的理解。本文将详细记录这个问题的发现、排查和最终解决方案，并分享一些避坑经验。

我们团队在开发一个订单处理系统时，使用了 Kafka 作为消息传递的中间件。系统中有多个消费者实例，它们都属于同一个消费者组（Consumer Group），用于并行消费订单消息。然而，在某个时间段内，部分消费者无法正常消费消息，导致订单处理延迟甚至丢失。

首先，我怀疑是消费者配置问题。Kafka 的消费者组机制决定了每个分区只能被组内的一个消费者消费。如果消费者组中的消费者数量大于分区数，那么部分消费者会处于空闲状态。但在我们的场景中，消费者数量是合理的，且分区数也足够。

接下来，我检查了消费者的启动逻辑。消费者是否正确地初始化了偏移量？是否有手动提交偏移量？这些都有可能影响消费行为。

此外，我还考虑了网络或 Kafka 集群本身的问题。例如，消费者是否能正常连接到 Kafka 集群？是否有心跳超时？这些都可能导致消费者被踢出消费者组。

首先，我查看了消费者的日志，发现某些消费者在启动后并没有接收到任何消息。日志中也没有明显的错误信息，这让我感到困惑。

INFO 2024-04-05 10:00:00,000 - Consumer started...
INFO 2024-04-05 10:00:01,000 - No messages received in the last 10 seconds.

2. 使用 Kafka 工具查看消费者组状态

运行以下命令查看消费者组的状态：

./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group

GROUP           | TOPIC        | PARTITION | CURRENT-OFFSET | LOG-END-OFFSET | LAG  | CONSUMER-ID      | HOST       | CLIENT-ID
my-group       | orders-topic | 0         | 100            | 120            | 20   | consumer-1       | /127.0.0.1  | consumer-1
my-group       | orders-topic | 1         | 80             | 100            | 20   | consumer-2       | /127.0.0.1  | consumer-2
my-group       | orders-topic | 2         | 0              | 100            | 100  | consumer-3       | /127.0.0.1  | consumer-3

可以看到，consumer-3 的偏移量为 0，而 log-end-offset 是 100，说明它还没有消费任何消息。

我们使用的是 Java 编写的 Kafka 消费者，以下是关键配置代码：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("enable.auto.commit", "false");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");

，意味着消费者不会自动提交偏移量。如果我们没有手动提交，那么即使消费者已经处理了消息，Kafka 也不会知道这个偏移量，从而导致重复消费或消费失败。

进一步检查消费者代码，发现消费者在处理消息时并未调用 

 方法。这意味着消费者虽然接收到了消息，但没有提交偏移量，导致 Kafka 认为这些消息尚未被消费。

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理消息逻辑
        System.out.println("Received message: " + record.value());
    }
}

这就是问题所在！由于没有提交偏移量，消费者在重启或重新加入消费者组时，会从最早的偏移量开始消费，导致部分消息被重复消费或未能及时消费。

修改消费者代码，在处理完每条消息后手动提交偏移量：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理消息逻辑
        System.out.println("Received message: " + record.value());
    }
    consumer.commitSync(); // 手动提交偏移量
}

这样，消费者就能在处理完一批消息后，主动提交偏移量，确保下次拉取时能够继续从上次的位置开始消费。

这次 Kafka 消费者组消费异常的问题，表面上看起来是一个简单的偏移量未提交的问题，但实际上涉及到了 Kafka 的消费者组机制、偏移量管理以及消费者逻辑设计等多个方面。通过排查和调试，我更加深入地理解了 Kafka 的工作原理，同时也积累了一些宝贵的实战经验。

在实际开发中，一定要注意消费者是否正确提交偏移量，尤其是在使用 

 的情况下。此外，合理设置消费者数量和分区数，也是保证系统稳定运行的重要因素。

Kafka消费者组消费异常的排查与解决实践

前端

后端

Kafka消费者组消费异常问题排查：订单处理系统中部分消费者无法正常消费消息，发现是由于未手动提交偏移量导致。通过检查消费者日志、Kafka工具和代码逻辑，最终修复方案是手动调用commitSync()提交偏移量。本文详细记录Kafka消费机制问题排查过程，分享偏移量管理和消费者组配置的实战经验。

消息队列

中间件

Kafka

Java

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

Kafka消费者组消费异常的排查与解决实践

Kafka消费者组消费异常的排查与解决实践

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐