Spring Boot嵌入式Kafka测试求助：@KafkaListener未接收消息

我之前也碰到过几乎一模一样的问题，咱们一步步排查和解决：

1. 先核对核心配置 #

• 确保嵌入式Kafka地址正确绑定：你的spring.kafka.bootstrap-servers必须指向嵌入式实例的地址，测试环境里建议用${spring.embedded.kafka.brokers}动态获取，比如在配置文件里写：

  spring:
    kafka:
      bootstrap-servers: ${spring.embedded.kafka.brokers}
  

  别硬编码localhost:9092，嵌入式Kafka可能会随机分配端口。
• 消费者偏移量配置：一定要把spring.kafka.consumer.auto-offset-reset设为earliest，不然消费者启动后会从最新偏移量开始，发在它启动前的消息就收不到了。
• 序列化/反序列化匹配：生产者的value-serializer和消费者的value-deserializer必须一致，比如都是StringSerializer/StringDeserializer，或者都是Json序列化器，否则消息会因为反序列化失败被静默丢弃（不会报错，但就是收不到）。

2. 检查@KafkaListener的写法 #

• 明确指定主题和GroupId：同一类里的多个Listener，建议给每个都单独指定groupId，避免默认GroupId导致的订阅冲突。比如：

  @KafkaListener(topics = "topic-a", groupId = "group-a")
  public void listenTopicA(String msg) { /* 处理逻辑 */ }
  
  @KafkaListener(topics = "topic-b", groupId = "group-b")
  public void listenTopicB(String msg) { /* 处理逻辑 */ }
  

  同时要确保发送消息时的主题名和Listener监听的主题名完全一致（大小写敏感）。
• Listener方法参数合法性：如果直接接收消息体，要确保参数类型和序列化后的类型匹配；如果用ConsumerRecord，要指定泛型类型，比如ConsumerRecord<String, YourDto>。

3. 测试代码的关键细节 #

• 用CountDownLatch等待消息处理：嵌入式Kafka的消息传递是异步的，别发完消息就直接断言，一定要用同步等待机制。比如在Listener类里定义CountDownLatch，收到消息后调用countDown()，测试代码里调用latch.await(5, TimeUnit.SECONDS)等待结果。
• 验证消息发送是否成功：给KafkaTemplate的发送结果加回调，确认消息没有发送失败：

  ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("topic-a", "test msg");
  future.addCallback(
      success -> System.out.println("消息发送成功：" + success.getRecordMetadata()),
      fail -> System.err.println("消息发送失败：" + fail.getMessage())
  );
  

4. 测试类注解要齐全 #

• 测试类必须同时加@SpringBootTest和@EmbeddedKafka，并且提前声明要用到的主题，比如：

  @SpringBootTest
  @EmbeddedKafka(topics = {"topic-a", "topic-b"}, partitions = 1)
  public class KafkaMultiListenerTest { /* ... */ }
  

  提前创建主题能避免“发送时才创建主题，Listener还没完成订阅”的问题。

5. 开启DEBUG日志排查细节 #

如果以上都没问题，就把Kafka相关日志调到DEBUG级别，查看消费者订阅、消息拉取的细节：

logging:
  level:
    org.springframework.kafka: DEBUG
    org.apache.kafka: DEBUG

你能看到消费者是否成功加入群组、是否拉取到消息、有没有反序列化异常等关键信息。

参考示例代码 #

给你一个能跑通的极简示例，对照着检查你的代码：

// 测试类
@SpringBootTest
@EmbeddedKafka(topics = {"test-topic-1", "test-topic-2"}, partitions = 1)
public class KafkaListenerTest {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private MultiTopicListeners listeners;

    @Test
    void testMultiListeners() throws InterruptedException {
        // 发送消息
        kafkaTemplate.send("test-topic-1", "Hello Topic 1");
        kafkaTemplate.send("test-topic-2", "Hello Topic 2");

        // 等待消息被处理
        listeners.getLatch1().await(5, TimeUnit.SECONDS);
        listeners.getLatch2().await(5, TimeUnit.SECONDS);

        // 断言结果
        assertEquals("Hello Topic 1", listeners.getMsg1());
        assertEquals("Hello Topic 2", listeners.getMsg2());
    }
}

// 多Listener组件
@Component
class MultiTopicListeners {
    private final CountDownLatch latch1 = new CountDownLatch(1);
    private final CountDownLatch latch2 = new CountDownLatch(1);
    private String msg1;
    private String msg2;

    @KafkaListener(topics = "test-topic-1", groupId = "group-1")
    public void listenTopic1(String message) {
        this.msg1 = message;
        latch1.countDown();
    }

    @KafkaListener(topics = "test-topic-2", groupId = "group-2")
    public void listenTopic2(String message) {
        this.msg2 = message;
        latch2.countDown();
    }

    // Getter方法
    public CountDownLatch getLatch1() { return latch1; }
    public CountDownLatch getLatch2() { return latch2; }
    public String getMsg1() { return msg1; }
    public String getMsg2() { return msg2; }
}

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