Spring中Reactive Kafka集成测试及消费者Mock实现方案咨询

方案一：复用@EmbeddedKafka配合Reactive Kafka客户端 #

@EmbeddedKafka并非只支持阻塞式客户端，它本质是启动一个嵌入式Kafka broker，完全可以和Reactive Kafka组件配合使用。你只需要在测试类上标注@EmbeddedKafka，然后配置ReactiveKafkaConsumerTemplate连接到这个嵌入式broker即可。

示例代码：

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = "test-topic")
class ReactiveKafkaConsumerTest {

    @Autowired
    private ReactiveKafkaConsumerTemplate<String, String> consumerTemplate;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate; // 用于发送测试消息

    @Test
    void testConsumerReceivesMessage() {
        // 发送测试消息
        kafkaTemplate.send("test-topic", "key", "test-message").block();

        // 订阅并验证消息
        Flux<ReceiverRecord<String, String>> messageFlux = consumerTemplate.receiveAutoAck();
        StepVerifier.create(messageFlux)
                .expectNextMatches(record -> record.value().equals("test-message"))
                .thenCancel()
                .verify(Duration.ofSeconds(5));
    }
}

注意：Spring Boot会自动注入@EmbeddedKafka的broker地址到spring.kafka.bootstrap-servers属性，无需手动配置。

方案二：直接Mock ReactiveKafkaConsumerTemplate #

如果不需要真实的Kafka broker，直接Mock消费者模板类，用Reactor的Flux模拟消息流即可。

示例代码：

@ExtendWith(MockitoExtension.class)
class ReactiveKafkaConsumerMockTest {

    @Mock
    private ReactiveKafkaConsumerTemplate<String, String> mockConsumerTemplate;

    @InjectMocks
    private YourReactiveConsumerService consumerService; // 你的消费者服务类

    @Test
    void testConsumerLogic() {
        // 模拟返回的消息流
        ReceiverRecord<String, String> mockRecord = ReceiverRecord.create(
                new ConsumerRecord<>("test-topic", 0, 0L, "key", "mock-message"),
                mock(Consumer.class)
        );
        when(mockConsumerTemplate.receiveAutoAck()).thenReturn(Flux.just(mockRecord));

        // 调用服务并验证逻辑
        Mono<String> result = consumerService.processMessages().next();
        StepVerifier.create(result)
                .expectNext("processed-mock-message") // 假设你的服务会处理消息并返回对应结果
                .verifyComplete();
    }
}

方案三：用Testcontainers启动真实Kafka容器 #

如果需要更贴近生产环境的测试，可以用Testcontainers启动真实的Kafka容器，配合Reactive Kafka客户端进行测试。

示例代码：

@SpringBootTest
@Testcontainers
class ReactiveKafkaTestcontainerTest {

    @Container
    static KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"));

    @DynamicPropertySource
    static void kafkaProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers);
    }

    @Autowired
    private ReactiveKafkaConsumerTemplate<String, String> consumerTemplate;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Test
    void testRealKafkaConsumer() {
        kafkaTemplate.send("test-topic", "key", "container-message").block();

        Flux<ReceiverRecord<String, String>> messageFlux = consumerTemplate.receiveAutoAck();
        StepVerifier.create(messageFlux)
                .expectNextMatches(record -> record.value().equals("container-message"))
                .thenCancel()
                .verify(Duration.ofSeconds(10));
    }
}

以上三种方案分别对应不同测试场景：快速单元测试用Mock，轻量集成测试用@EmbeddedKafka，贴近生产环境的测试用Testcontainers。

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