如何在 JUnit 集成测试中立即启动 Spring Integration poller

2026-10-10支付小周

结论

建议在测试中禁止入站端点自动启动，再通过 MockIntegrationContext.substituteTriggerFor() 把正式的定时触发器替换为 OnlyOnceTrigger，最后手动启动端点。

这种写法有几个好处：

保留真实的 ReadTimeoutsMessageSource，可以验证存储过程的结果是否被正确构造成消息；
轮询会立即执行一次，不必等待 10 秒；
不会因固定周期轮询而重复调用，测试也更稳定；
可以覆盖从 .split() 到最终 handler 的完整消息流。

如果只想测试下游 flow，不关心原始 MessageSource 如何构造消息，更适合使用 substituteMessageSourceFor()。

测试写法

先确认测试依赖中包含 spring-integration-test。测试类可以这样配置：

@SpringBootTest
@SpringIntegrationTest(noAutoStartup = "messageTimeout.readTimeouts")
class MessageTimeoutIntegrationTest {

    @Autowired
    private MockIntegrationContext mockIntegrationContext;

    @Autowired
    private ApplicationContext applicationContext;

    @MockBean
    private UpdatePaymentWithStatusTimeoutHandler handler;

    @AfterEach
    void tearDown() {
        applicationContext
                .getBean(
                        "messageTimeout.readTimeouts",
                        SourcePollingChannelAdapter.class
                )
                .stop();

        mockIntegrationContext.resetBeans();
    }

    @Test
    void shouldConsumeEveryTimeoutReturnedByStoredProcedure() {
        List<TimeoutDto> rows = List.of(
                new TimeoutDto(/* ... */),
                new TimeoutDto(/* ... */),
                new TimeoutDto(/* ... */)
        );

        // 根据 ReadTimeoutsMessageSource 的实际依赖调整此处。
        when(storedProcedure.execute()).thenReturn(rows);

        mockIntegrationContext.substituteTriggerFor(
                "messageTimeout.readTimeouts",
                new OnlyOnceTrigger()
        );

        SourcePollingChannelAdapter endpoint =
                applicationContext.getBean(
                        "messageTimeout.readTimeouts",
                        SourcePollingChannelAdapter.class
                );

        endpoint.start();

        verify(handler, timeout(2_000).times(rows.size()))
                .handle(any());
    }
}

为什么要禁止自动启动

如果端点跟随 Spring 容器自动启动，正式的 fixedDelay 触发器可能已经进入调度。此时测试再替换触发器，就会出现时序竞争，结果可能取决于测试机器的执行速度。

@SpringIntegrationTest(
        noAutoStartup = "messageTimeout.readTimeouts"
)

这样可以确保测试按以下顺序执行：

Spring 创建 flow，但不启动轮询端点；
mock 存储过程的返回值；
使用 OnlyOnceTrigger 替换正式触发器；
手动启动端点；
端点立即轮询一次；
MessageSource 返回消息；
.split() 拆分消息；
最终 handler 逐项消费。

OnlyOnceTrigger 会在首次调度时立即触发，之后不再提供下一次执行时间，因此不会继续周期性查询。

异步断言不要使用固定休眠

轮询端点通常由调度线程执行。调用 endpoint.start() 后，不应马上使用普通的 verify(handler).handle(...)，也不建议这样写：

Thread.sleep(1000);

可以改用 Mockito 的超时验证：

verify(handler, timeout(2_000).times(rows.size()))
        .handle(any());

如果还要检查每条消息的具体内容，可以配合 ArgumentCaptor：

ArgumentCaptor<TimeoutDto> captor =
        ArgumentCaptor.forClass(TimeoutDto.class);

verify(handler, timeout(2_000).times(rows.size()))
        .handle(captor.capture());

assertThat(captor.getAllValues())
        .containsExactlyInAnyOrderElementsOf(rows);

如果 handler 接收的是 Message<TimeoutDto>，则捕获 Message<?>，再检查 payload 和所需的 headers。

何时使用 `substituteMessageSourceFor`

如果只需要验证 .split()、日志处理和最终 handler，不准备测试 ReadTimeoutsMessageSource，可以直接替换消息源：

MessageSource<?> testSource = () ->
        MessageBuilder
                .withPayload(List.of(row1, row2, row3))
                .build();

mockIntegrationContext.substituteMessageSourceFor(
        "messageTimeout.readTimeouts",
        testSource
);

mockIntegrationContext.substituteTriggerFor(
        "messageTimeout.readTimeouts",
        new OnlyOnceTrigger()
);

applicationContext
        .getBean(
                "messageTimeout.readTimeouts",
                SourcePollingChannelAdapter.class
        )
        .start();

这几种写法的测试范围不同：

保留原始 MessageSource，只替换 trigger：验证数据库调用的 mock 结果、消息构造、拆分和最终消费，适合当前场景。
同时替换 MessageSource 和 trigger：只验证消息进入 flow 后的处理过程，测试范围更小。
直接向内部 channel 发送消息：适合单独测试下游组件，但需要为相应 channel 明确命名，而且会绕过入站适配器和 MessageSource。

注意事项

端点名称必须与 DSL 中配置的 ID 一致：

.id("messageTimeout.readTimeouts")

因此，无论是替换 trigger、禁止自动启动，还是获取端点，都应使用：

messageTimeout.readTimeouts

测试结束后，还应停止手动启动的端点并调用 mockIntegrationContext.resetBeans()，避免替换后的测试组件或运行状态影响同一 Spring 测试上下文中的其他测试。

不同 Spring Integration 版本对 noAutoStartup 的名称匹配方式或端点 Bean 的暴露方式可能有所不同，具体应以项目所用版本的 API 为准。整体做法不变：先阻止自动调度，再用 OnlyOnceTrigger 替换正式 trigger，最后手动启动 SourcePollingChannelAdapter。