异步Spring Integration全量流执行完成的回调通知实现方案咨询

针对多层ExecutorChannel并行处理无法感知全流程完成的问题，以下是3种可落地的实现方案，按推荐优先级排序：

方案1：使用Spring Integration原生Aggregator（推荐） #

这是Spring Integration官方针对异步分支聚合场景的标准实现，核心逻辑是分层聚合异步任务：先聚合第二层executorChannelTwo的子任务，再聚合第一层executorChannelOne的父任务，所有任务完成后触发回调。

改造步骤： #

1. 给所有消息增加批次标识头
   首先改造消息生成逻辑，给每个批次生成唯一batchId，同时携带对应层级的任务总数：

@Component
@Scope("prototype")
public class ServiceHandlerOne{
    @Autowired
    MessageChannel executorChannelOne;
    
    @ServiceActivator
    public void handle(Message<?> message) {
        // 生成当前批次唯一标识
        String batchId = UUID.randomUUID().toString();
        List<?> parties = repository.findAll("parties");
        int partyTotal = parties.size();
        // 给每个party消息增加批次头、总party数
        parties.forEach(party -> {
            Message<?> msg = MessageBuilder.withPayload(party)
                    .setHeader("batchId", batchId)
                    .setHeader("partyTotal", partyTotal)
                    .build();
            executorChannelOne.send(msg);
        });
    }
}

@Component
@Scope("prototype")
public class ParallelServiceHandlerOne{
    @Autowired
    MessageChannel executorChannelTwo;
    
    @ServiceActivator
    public void handle(Message<?> message) {
        String batchId = message.getHeaders().get("batchId", String.class);
        // 假设party对象有唯一标识字段id
        Object partyId = ((Party)message.getPayload()).getId();
        List<?> subDatas = repository.findAll("party");
        int subTaskTotal = subDatas.size();
        // 给子任务消息增加party标识、子任务总数
        subDatas.forEach(data -> {
            Message<?> msg = MessageBuilder.withPayload(data)
                    .setHeader("batchId", batchId)
                    .setHeader("partyId", partyId)
                    .setHeader("subTaskTotal", subTaskTotal)
                    .build();
            executorChannelTwo.send(msg);
        });
    }
}

2. 新增分层聚合流

// 第一层：聚合单个party对应的所有子任务
@Bean
public IntegrationFlow subTaskAggregateFlow() {
    return IntegrationFlows.from("executorChannelTwo")
            // 先执行原有的子任务处理逻辑
            .handle("parallelServiceHandlerTwo", "handle")
            // 按partyId聚合同属一个party的子任务
            .aggregate(aggregatorSpec -> aggregatorSpec
                    .correlationExpression("headers.partyId")
                    // 子任务完成数等于总数时释放聚合结果
                    .releaseExpression("size() == headers.subTaskTotal[0]")
                    // 聚合完成后发送到party完成通道
                    .outputChannel("partyCompleteChannel")
                    // 可选：配置聚合超时时间，避免异常任务导致批次永久挂起
                    .groupTimeout(3600000)
            )
            .get();
}

// 第二层：聚合当前批次所有party的完成通知
@Bean
public IntegrationFlow partyAggregateFlow() {
    return IntegrationFlows.from("partyCompleteChannel")
            .aggregate(aggregatorSpec -> aggregatorSpec
                    .correlationExpression("headers.batchId")
                    // 完成的party数等于总party数时释放
                    .releaseExpression("size() == headers.partyTotal[0]")
                    .outputChannel("batchCompleteChannel")
                    .groupTimeout(7200000)
            )
            .get();
}

// 全批次完成回调
@Bean
public IntegrationFlow batchCompleteFlow() {
    return IntegrationFlows.from("batchCompleteChannel")
            .handle(msg -> {
                String batchId = msg.getHeaders().get("batchId", String.class);
                // 此处编写全流程完成后的业务逻辑，如发通知、更新任务状态、记统计日志
                log.info("批次{}全流程处理完成", batchId);
            })
            .nullChannel();
}

方案2：自定义计数拦截器（轻量改造） #

如果不想调整现有流结构，可以通过通道拦截器+原子计数器实现任务统计：

1. 自定义通道拦截器统计任务发送和完成数量：

public class BatchTaskInterceptor implements ChannelInterceptor {
    // 存储每个批次待完成的任务数
    private final ConcurrentHashMap<String, AtomicInteger> pendingTaskCounter = new ConcurrentHashMap<>();
    // 存储每个批次的完成回调
    private final ConcurrentHashMap<String, Runnable> completeCallbackMap = new ConcurrentHashMap<>();

    // 任务发送前计数+1
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        String batchId = message.getHeaders().get("batchId", String.class);
        if (batchId != null) {
            pendingTaskCounter.computeIfAbsent(batchId, k -> new AtomicInteger(0)).incrementAndGet();
        }
        return message;
    }

    // 任务处理完成后计数-1，减到0触发回调
    public void afterTaskComplete(String batchId) {
        AtomicInteger counter = pendingTaskCounter.get(batchId);
        if (counter == null) return;
        int remain = counter.decrementAndGet();
        if (remain == 0) {
            completeCallbackMap.get(batchId).run();
            pendingTaskCounter.remove(batchId);
            completeCallbackMap.remove(batchId);
        }
    }

    public void registerBatchCallback(String batchId, Runnable callback) {
        completeCallbackMap.put(batchId, callback);
    }
}

2. 给两个ExecutorChannel注册拦截器，同时在parallelServiceHandlerTwo的处理逻辑最后调用afterTaskComplete方法即可。

方案3：线程池状态监控（仅适合测试/非核心场景） #

如果不需要精确的完成通知，只是日常调度监控用，可以定期检查两个ExecutorChannel对应的线程池状态：当活跃线程数为0且队列任务数为0时，判定当前批次处理完成，该方案准确性较低，不建议生产环境使用。

注意事项 #

• 所有方案都建议配置超时兜底逻辑，避免个别任务处理失败导致批次状态永久挂起
• 生产环境任务量较大时，方案1的聚合器可以配置JDBC消息存储，避免应用重启丢失聚合进度
• 任务处理异常时可以配置重试逻辑，或者聚合器配置partial-release策略，完成后上报异常任务清单

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