如何让动态注册的多个IntegrationFlow在独立线程运行并写入同一输出通道？

首先纠正一个关键误解：Spring Integration中Poller的taskExecutor()配置，会负责整个轮询任务的线程执行，包括调用JdbcPollingChannelAdapter的receive()方法拉取数据，以及后续的消息处理流程，并非只处理轮询之后的步骤。要实现每个动态注册的IntegrationFlow在独立线程运行，核心是为每个Flow的Poller分配专属的线程资源。

具体实现步骤 #

1. 为每个动态注册的Flow创建独立的TaskExecutor，确保每个Flow的轮询和处理逻辑在专属线程中执行。
2. 所有Flow统一输出到同一个outChannel。

代码示例 #

private IntegrationFlowContext context;

// 动态注册多个独立线程的Flow
context.registration(createFlowWithUniqueThread()).register();
context.registration(createFlowWithUniqueThread()).register();

private IntegrationFlow createFlowWithUniqueThread() {
    // 为每个Flow创建专属的单线程线程池
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(1);
    taskExecutor.setMaxPoolSize(1);
    taskExecutor.setThreadNamePrefix("flow-worker-");
    taskExecutor.initialize();

    return IntegrationFlow.from(messageSource(), 
                    p -> p.poller(Pollers.fixedDelay(5000)
                            .taskExecutor(taskExecutor))) // 绑定专属线程池
            .channel("outChannel")
            .get();
}

private MessageSource<Object> messageSource() {
    JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(dataSource, 
            "SELECT * FROM your_target_table");
    // 配置适配器的行映射、参数等属性
    // adapter.setRowMapper(...);
    return adapter;
}

关键说明 #

• 每个动态注册的Flow都会初始化一个单线程ThreadPoolTaskExecutor，这样该Flow的轮询触发、数据库查询、消息发送全流程都会在这个专属线程中运行，完全隔离其他Flow的线程资源。
• 线程池的threadNamePrefix可以帮助你在日志中区分不同Flow的线程，方便排查问题。
• 如果需要更高的并发度，可以调整线程池的corePoolSize和maxPoolSize，但每个Flow仍需使用独立的线程池实例，确保线程资源隔离。

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