Kafka消费者无法接收生产者消息（附Spring Boot代码求助）

我先是按照相关教程搭建了Kafka控制台生产者和消费者，但消费者始终接收不到生产者发送的消息。之后改用Spring Boot实现Kafka消费者和生产者，问题依然存在。我已经查阅了Stack Overflow上的相关帖子，但都没能解决问题。

相关代码 #

Application.java 中的配置Bean #

@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> kafkaConsumerFactory,
        KafkaTemplate<Object, Object> template) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory);
    factory.setErrorHandler(new SeekToCurrentErrorHandler(
            new DeadLetterPublishingRecoverer(template), 3)); // dead-letter after 3 tries
    return factory;
}

Consumer.java 中的监听方法 #

@KafkaListener(id = "com.stellapps.rtcreset.kafka.consumer", topics = "${rawEventsTopicName}") // add it in config part
public void messageListener(String in) {
    System.out.println("Received from message: " + in);
    rtcInterpreter.call(in,url);
}

排查建议 #

我整理了几个常见的排查方向，你可以逐一验证：

• 确认基础配置是否正确

  • 检查application.properties或application.yml中的Kafka核心配置：
    • spring.kafka.bootstrap-servers是否指向正确的Kafka Broker地址，默认端口9092，要确保没有被防火墙拦截
    • 消费者的group.id是否配置（如果没显式设置，Spring Boot会用默认值，但建议显式指定）
    • rawEventsTopicName对应的主题是否真实存在？可以用Kafka命令行工具kafka-topics.sh --list --bootstrap-server <broker地址>确认
  • 生产者与消费者必须使用同一个bootstrap-servers配置，否则跨集群肯定无法互通

• 主题与消息的兼容性检查

  • 消息序列化/反序列化是否匹配：你消费者接收的是String类型，要确保生产者发送的是字符串格式，同时消费者的spring.kafka.consumer.value-deserializer设置为org.apache.kafka.common.serialization.StringDeserializer
  • 检查主题的健康状态：用kafka-topics.sh --describe --topic <你的主题名> --bootstrap-server <broker地址>查看，确保分区的Leader、副本、ISR列表都处于正常状态

• Spring Boot Kafka的特定检查

  • 启动类是否添加了@EnableKafka注解？这是开启Kafka监听功能的必要条件，没加的话@KafkaListener注解完全不会生效
  • 检查消费者GroupId是否重复：如果同一个GroupId下有其他未关闭的消费者进程，消息会被负载均衡到其他进程，导致当前服务收不到
  • 开启DEBUG日志排查：在配置文件中把logging.level.org.springframework.kafka设为DEBUG，查看容器是否正常启动、是否成功连接Broker、是否订阅到目标主题

• 用控制台工具验证链路

  • 用控制台生产者发送消息，同时用控制台消费者（指定和Spring消费者相同的GroupId）接收，测试链路是否通畅：
    生产者命令：kafka-console-producer.sh --broker-list <broker地址> --topic <你的主题名>
    消费者命令：kafka-console-consumer.sh --bootstrap-server <broker地址> --topic <你的主题名> --group <你的消费者GroupId> --from-beginning
    如果控制台消费者也收不到，问题大概率出在Kafka集群或主题配置上；如果控制台能收到，那就是Spring Boot配置的细节问题

内容的提问来源于stack exchange，提问作者user11311618