Python KafkaConsumer消费异常：消息时而可接收时而无法接收

问题现象 #

Spring Boot后端向指定Kafka主题pas-advance-message发送消息时，使用KafkaConsumer库的Python客户端无法稳定接收消息——时而能收到，时而完全收不到。但通过Kafka控制台消费者命令：

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic pas-advance-message

可以正常看到所有消息。

相关代码 #

Python消费者代码（main.py） #

while True:
    messages = consumer.poll(timeout_ms=1000)

    print(f"There is no message  : {messages}")

    for topic_partition, message_list in messages.items():
        for message in message_list:
            if message is None:
                continue
            else:
                value = message.value.decode('utf-8')  # 假设消息使用utf-8编码
                kafka_arr = value.split(',')
                value = kafka_arr[0]
                stationCode = kafka_arr[1]
                print(f"{stationCode} stationCode comes from api {getStation} comes from current directory")

                if stationCode == getStation:
                    print('Match - Station')
                    switch_case(value)
                else:
                    continue

Spring Boot生产者代码 #

private final KafkaTemplate<String, String> kafkaTemplate;
private static final String TOPIC_NAME = "pas-advance-message";

@Autowired
public KafkaProducerService(KafkaTemplate<String, String> kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
}

public void sendMessage(String message) {
    kafkaTemplate.send(TOPIC_NAME, message);
}

排查与解决建议 #

1. Group ID与偏移量问题 #

• 若Python消费者使用的Group ID之前消费过该主题，Kafka会留存该Group的消费偏移量。如果偏移量已处于主题最新位置，新启动的消费者无法接收历史消息；若偏移量提交不及时，会导致消息重复接收或丢失，表现为"时而收到时而收不到"。
• 解决方式：
  • 显式控制偏移量提交：将enable_auto_commit设为False，在处理完消息后手动调用consumer.commit()，避免自动提交的不确定性。
  • 测试阶段使用全新的Group ID，规避旧偏移量的影响。
  • 不要随意"清除Group ID"，而是通过kafka-consumer-groups.sh脚本重置指定Group的偏移量到最早或最新位置。

2. Python消费者配置优化 #

• 确认bootstrap_servers配置与生产者、控制台消费者一致（确保都是localhost:9092）。
• 检查auto_offset_reset：若设置为latest，消费者启动后只会接收启动后的新消息；若需要接收历史消息，需改为earliest。
• 调整fetch_min_bytes和fetch_max_wait_ms：如果fetch_min_bytes值过高，Kafka会攒够指定字节数才返回消息，导致延迟；可适当降低fetch_min_bytes，或调整fetch_max_wait_ms（默认500ms）匹配你的poll(timeout_ms=1000)逻辑。

3. Spring Boot生产者可靠性验证 #

• 当前生产者代码未确认消息发送状态，建议添加回调机制，排查是否存在消息发送失败的情况：

  public void sendMessage(String message) {
      kafkaTemplate.send(TOPIC_NAME, message)
          .addCallback(success -> {
              if (success != null) {
                  System.out.println("消息发送成功：" + success.getRecordMetadata());
              }
          }, failure -> {
              System.err.println("消息发送失败：" + failure.getMessage());
          });
  }
  

• 调整生产者acks配置：默认acks=1仅等待Leader节点确认，若需更高可靠性，可设为acks=all（等待所有同步副本确认），避免消息未同步就丢失。

4. Kafka集群状态检查 #

• 先通过以下命令查看主题的分区、副本状态，确认所有副本处于同步状态：

  ./bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic pas-advance-message
  

• 非必要不要修改Kafka默认配置，若发现副本同步延迟、分区数不足等集群层面问题，再针对性调整。

内容的提问来源于stack exchange，提问作者Sanberk Küçükelepçe