在 Spring 集成流 DSL 中间实现轮询

Question

我正在编写一个 spring 集成 DSL 流程。 如下图所示。

正如您在流程中看到的，我需要从数据库中读取 1 百万个实体。我想避免一次性阅读这些内容。

我想实现轮询，它会以固定的间隔读取 N 个实体并将其发送以进行处理。

在我阅读的轮询示例中，轮询用作流程的第一步。就我而言，我想在流程中间实施。

请告诉我如何实现。

感谢任何帮助。 提前致谢。

Answer 1

如果您想使用一些外部刺激来触发一些轮询流的开始，请使用控制总线：

@SpringBootApplication
public class So63337649Application {

    public static void main(String[] args) {
        SpringApplication.run(So63337649Application.class, args);
    }

    @Bean
    IntegrationFlow trigger(ConnectionFactory connectionFactory) {
        return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "foo"))
                .transform(msg -> "@poller.start()")
                .channel("control.input")
                .get();
    }

    @Bean
    IntegrationFlow control() {
        return f -> f.controlBus();
    }

    @Bean
    IntegrationFlow mainFlow() {
        return IntegrationFlows.from(() -> "foo", e -> e
                    .id("poller")
                    .autoStartup(false)
                    .poller(Pollers.fixedDelay(5000)))
                .handle(System.out::println)
                .get();
    }

}