如何等待Spring Integration流执行完成后调用stop()再退出主类

解决方案 #

你可以通过CountDownLatch计数+等待处理完成的逻辑实现需求，具体修改如下：

步骤1：定义计数锁Bean

先统计目标目录下的待处理文件总数，初始化CountDownLatch的计数：

@Bean
public CountDownLatch processingCompleteLatch() throws IOException {
    File targetDir = messageFlow1Directory().getDirectory();
    long fileCount = Files.list(targetDir.toPath())
            .filter(Files::isRegularFile)
            .count();
    return new CountDownLatch((int) fileCount);
}

步骤2：修改IntegrationFlow配置，新增处理完成的计数逻辑

在publishSubscribeChannel的订阅最后新增一个处理节点，每完成一个文件的处理就将计数减1：

@Bean
public IntegrationFlow processMessage(CountDownLatch processingCompleteLatch) {
    return IntegrationFlows
            .from(messageFlow1Directory(), c -> c
                    .id("flow1")
                    .autoStartup(false)
                    .poller(Pollers.fixedDelay(30000)
                            // 单次轮询处理所有可用文件，避免分批次轮询
                            .maxMessagesPerPoll(-1)
                            .errorChannel(errorMessageChannel())))
            .publishSubscribeChannel(s -> s
                    .subscribe(f -> f.handle("flow1Service", "processMessage"))
                    .subscribe(f -> f.handle(otherAction))
                    // 新增：处理完成后计数减1
                    .subscribe(f -> f.handle(message -> processingCompleteLatch.countDown())))
            .get();
}

步骤3：修改主类逻辑，等待计数完成后停止流并退出

public static void main(String[] args) throws InterruptedException {
    ConfigurableApplicationContext context = SpringApplication.run(Flow1Main.class, args);
    CountDownLatch latch = context.getBean(CountDownLatch.class);
    MessageChannel controlBus = context.getBean("controlBus", MessageChannel.class);
    
    // 启动流
    controlBus.send(new GenericMessage<>("@flow1.start()"));
    // 阻塞等待所有文件处理完成
    latch.await();
    // 停止流
    controlBus.send(new GenericMessage<>("@flow1.stop()"));
    // 关闭Spring上下文后退出程序
    context.close();
}

注意事项 #

• 如果处理逻辑抛出异常，要在errorChannel对应的处理流程中也添加countDown逻辑，避免锁永久等待
• 如果你不需要持续定时轮询目录，只需要处理一次现有文件就退出，可以将轮询器改为一次性触发器，不需要配置30秒的固定延迟
• 如果你的场景是动态新增文件、需要满足特定终止条件才停止，可以将CountDownLatch替换为AtomicBoolean终止标志，主方法轮询该标志位即可

内容的提问来源于stack exchange，提问作者Brahim B.