Spring集成并发 - 检测完成答案

【问题标题】：Spring integration concurrency - detecting completionSpring集成并发 - 检测完成
【发布时间】：2021-04-27 03:43:04
【问题描述】：

我有一个 spring 集成工作流，它在其通道中嵌入了任务执行器，以便启用并发处理。我通过网关手动启动处理，并且需要阻塞主线程，直到所有异步进程都完成。有没有办法做到这一点？我尝试过沿着屏障、闩锁和通道拦截器的思路思考，但没有解决方案即将到来。有什么想法吗？

【问题讨论】：

标签： spring concurrency spring-integration terminate

【解决方案1】：

查看参考手册中的聚合器部分： http://static.springsource.org/spring-integration/docs/latest-ga/reference/htmlsingle/#aggregator

如果聚合器位于网关的下游，则网关调用者可以阻塞（或使用 Future，如果这被定义为网关接口上的返回类型），直到聚合器接收并释放相关的消息组，即使那些在不同的线程上异步处理。

本质上，Aggregator 本身就是一个屏障，它的默认释放策略本质上是一个基于消息组序列大小的倒计时锁存器。

希望对您有所帮助。 -马克

【讨论】：

嗨，马克，非常感谢。但是，我不在工作流程中使用聚合器。我使用拆分器，但只是对输入进行批处理（即从 150,000 个元素的集合中，我创建了 100 个元素的批次）。我想到的一种解决方案是扩展 ThreadPoolTaskExecutor 并根据已租用和待处理批次的线程数创建一个锁存器 - 当它们都降为零时，然后打开锁存器。不确定这是否可行，以及就优雅的解决方案而言，它是否是最好的。

【解决方案2】：

为了回答我自己的问题，这就是我最终要做的：

创建一个知道何时关闭的自定义 ExecutorService - 在我的情况下，这只是在释放最后一个活动线程时 - 即在执行工作流中的最后一个部分之后：


public class WorkflowThreadPoolExecutor extends ScheduledThreadPoolExecutor {
    public WorkflowThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize);
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);

        if (getActiveCount() == 1) {
            shutdown();
        }
    }
}

在主线程中等待执行程序终止，如下所示：


    try {
        executorService.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
    } catch (InterruptedException ex) {
        LOG.error("message=Error awaiting termination of executor", ex);
    }

希望这可以帮助其他面临类似问题的人。

【讨论】：