如何测试Spring Cloud Stream可轮询Supplier并触发调度器？

1. 依赖配置 #

确保测试类路径中引入spring-cloud-stream-test-binder依赖：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-test-binder</artifactId>
    <scope>test</scope>
</dependency>

2. 基础测试：手动触发消息生成 #

通过TestChannelBinderConfiguration绑定测试通道，直接调用Supplier的get()方法模拟调度触发，然后断言消息：

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.Message;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
@Import(TestChannelBinderConfiguration.class)
public class MyMessagesSupplierTest {

    @Autowired
    private OutputDestination outputDestination;

    @Autowired
    private MyMessagesSupplier myMessagesSupplier;

    @Test
    void testPollableSupplier() {
        // 手动调用get()，模拟调度器触发消息生成
        myMessagesSupplier.get().subscribe();

        // 从测试通道获取指定目标的消息
        Message<byte[]> receivedMessage = outputDestination.receive(1000, "myMessagesSupplier-out-0");

        // 断言消息内容
        assertThat(receivedMessage).isNotNull();
        assertThat(new String(receivedMessage.getPayload())).isEqualTo("My message");
    }
}

3. 进阶测试：触发Spring调度任务 #

如果需要验证调度器本身的逻辑，而非直接调用方法，可以通过ScheduledTaskHolder获取注册的调度任务并手动触发：

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.Message;
import org.springframework.scheduling.config.ScheduledTask;
import org.springframework.scheduling.config.ScheduledTaskHolder;

import java.util.Set;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
@Import(TestChannelBinderConfiguration.class)
public class MyMessagesSupplierSchedulerTest {

    @Autowired
    private OutputDestination outputDestination;

    @Autowired
    private ScheduledTaskHolder scheduledTaskHolder;

    @Test
    void testTriggerScheduler() {
        // 获取所有注册的调度任务
        Set<ScheduledTask> scheduledTasks = scheduledTaskHolder.getScheduledTasks();
        assertThat(scheduledTasks).isNotEmpty();

        // 触发对应PollableBean的调度任务
        scheduledTasks.stream()
                .findFirst()
                .ifPresent(task -> task.getTask().run());

        // 断言消息
        Message<byte[]> receivedMessage = outputDestination.receive(1000, "myMessagesSupplier-out-0");
        assertThat(receivedMessage).isNotNull();
        assertThat(new String(receivedMessage.getPayload())).isEqualTo("My message");
    }
}

关键说明 #

• TestChannelBinderConfiguration会替换生产环境binder为测试实现，将消息路由到OutputDestination，方便直接断言。
• 手动调用get()适合快速验证消息生成逻辑；触发调度任务则更贴近真实运行场景，可用于验证调度配置是否生效。
• 测试时可临时覆盖轮询间隔配置（如spring.cloud.stream.bindings.myMessagesSupplier-out-0.producer.poller.fixed-delay=100），但手动触发通常更高效。

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