在Split/Aggregate集成流中使用ExpressionEvaluatingRequestHandlerAdvice遇异常处理问题

1. 核心配置要点 #

（1）为文件写入处理器配置正确的ExpressionEvaluatingRequestHandlerAdvice #

需要确保Advice在捕获异常后保留原消息的headers（尤其是replyChannel），返回指定的false结果，同时不中断流程：

@Bean
public ExpressionEvaluatingRequestHandlerAdvice fileWriteErrorAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    // 异常发生时返回false
    advice.setReturnFailureExpressionString("false");
    // 捕获异常并返回结果，而非抛出异常
    advice.setTrapException(true);
    // 保留原消息headers，确保聚合所需的replyChannel不丢失
    advice.setCopyHeaders(true);
    return advice;
}

（2）Split/Aggregate流的完整配置 #

Split阶段必须启用applySequence(true)，确保拆分后的消息携带聚合所需的correlationId、sequenceNumber等headers；聚合阶段则收集所有成功/失败结果：

@Bean
public IntegrationFlow fileBatchWriteFlow() {
    return f -> f
            // 输入：待写入的文件数据列表
            .split()
            .applySequence(true)
            // 绑定异常处理Advice到文件写入处理器
            .handle(fileWritingHandler(), e -> e.advice(fileWriteErrorAdvice()))
            // 聚合所有结果，返回包含true/false的列表
            .aggregate(a -> a
                    .outputProcessor(group -> group.getMessages()
                            .stream()
                            .map(Message::getPayload)
                            .collect(Collectors.toList()))
                    .correlationStrategy(m -> m.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID))
                    .releaseStrategy(group -> group.getMessages().size() == group.getSequenceSize()));
}

（3）文件写入处理器配置 #

确保处理器会返回成功标识，且异常能被Advice捕获：

@Bean
public MessageHandler fileWritingHandler() {
    FileWritingMessageHandler handler = new FileWritingMessageHandler(new File("/target/directory"));
    handler.setFileExistsMode(FileExistsMode.REPLACE);
    handler.setExpectReply(true); // 必须启用回复，确保返回结果
    // 将写入成功的结果转换为true
    handler.setReplyPayloadExpression(new LiteralExpression("true"));
    return handler;
}

2. 解决DestinationResolutionException的关键 #

• 保留replyChannel header：Advice设置copyHeaders(true)，保证异常处理后的消息仍带有Split阶段生成的动态临时回复通道，让结果能正确路由到聚合器。
• 启用setTrapException(true)：阻止异常抛出，改用指定结果继续流程，避免中断。
• 处理器开启expectReply(true)：确保无论成功或失败，都有明确的payload传递给聚合器。

3. 常见误区修正 #

• 不要手动给Advice指定output-channel：Split后的replyChannel是动态生成的临时通道，手动指定会覆盖该header，导致聚合器无法接收结果。
• 必须启用applySequence(true)：否则聚合器无法识别拆分后的消息属于同一组，无法完成结果汇总。

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