如何使用Spring Embedded Kafka测试@KafkaListener

嘿，我来帮你搞定这个Spring Boot Kafka监听器的单元测试！用Spring Embedded Kafka绝对是正确的选择——不用折腾真实的Kafka集群和ZooKeeper，测试起来高效又省心。结合你给出的Listener代码，我给你整理一套完整的测试方案：

1. 先搞定依赖 #

首先要确保你的测试依赖里包含spring-kafka-test，Spring Boot的spring-boot-starter-test其实已经间接包含了它，但为了明确，你可以在构建文件里显式添加：

Maven 依赖 #

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>

Gradle 依赖 #

testImplementation 'org.springframework.kafka:spring-kafka-test'

2. 编写单元测试类 #

接下来写测试代码，我们用@SpringBootTest加载Spring上下文，@EmbeddedKafka自动启动嵌入式Kafka，然后用KafkaTemplate发送测试消息，最后通过CountDownLatch验证监听器是否收到消息：

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.context.EmbeddedKafka;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertTrue;

@SpringBootTest
// 指定要创建的topic，分区数设为1足够测试用
@EmbeddedKafka(topics = "sample-topic", partitions = 1)
public class ListenerTest {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private CountDownLatch latch;

    @Test
    void testKafkaListenerReceivesMessage() throws InterruptedException {
        // 发送一条测试消息到目标topic
        kafkaTemplate.send("sample-topic", "hello-test");

        // 等待latch计数归零，超时时间设为5秒，避免无限等待
        boolean isMessageReceived = latch.await(5, TimeUnit.SECONDS);

        // 断言监听器确实收到了消息
        assertTrue(isMessageReceived, "监听器未在指定时间内处理消息");
    }
}

3. 配置CountDownLatch Bean #

因为你的Listener依赖CountDownLatch，所以需要在测试配置里（或者主配置类，如果你希望全局可用的话）定义这个Bean：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.CountDownLatch;

@Configuration
public class TestKafkaConfig {

    @Bean
    public CountDownLatch countDownLatch() {
        // 计数设为1，因为我们只发送一条测试消息
        return new CountDownLatch(1);
    }
}

几个实用小提示 #

• @EmbeddedKafka支持很多自定义参数，比如brokerCount设置Broker数量，ports指定端口，根据你的测试场景调整即可。
• 如果你的监听器处理的是自定义对象（不是String），记得配置对应的序列化/反序列化器，测试时KafkaTemplate也要用对应的序列化器发送消息。
• Spring会自动把嵌入式Kafka的地址注入到spring.kafka.bootstrap-servers，不用手动配置，非常省心。

这样一套流程下来，你就能快速验证你的Kafka监听器逻辑是否正常，完全不用启动真实的Kafka集群！

内容的提问来源于stack exchange，提问作者riccardo.cardin