Spring Integration如何通过Advice同时返回Payload与异常？

核心思路 #

通过自定义结果封装DTO承载子消息的原始Payload与异常信息，结合ExpressionEvaluatingRequestHandlerAdvice捕获子处理阶段的异常并返回结构化结果，最终在聚合阶段统一处理所有成功/失败条目。

1. 定义结果封装DTO #

创建ProcessingResult类，用于单独存储每个子消息的处理状态、原始Payload和异常：

public class ProcessingResult<T> {
    private final T payload;
    private final Throwable exception;

    private ProcessingResult(T payload, Throwable exception) {
        this.payload = payload;
        this.exception = exception;
    }

    // 成功结果构造方法
    public static <T> ProcessingResult<T> success(T payload) {
        return new ProcessingResult<>(payload, null);
    }

    // 失败结果构造方法
    public static <T> ProcessingResult<T> failure(T payload, Throwable exception) {
        return new ProcessingResult<>(payload, exception);
    }

    // Getter方法
    public T getPayload() { return payload; }
    public Throwable getException() { return exception; }
    public boolean isSuccess() { return exception == null; }
}

2. 配置Integration Flow #

使用Java DSL配置拆分、子消息处理（带异常捕获）、聚合及统一结果处理的完整流程：

@Configuration
public class SplitAggregateErrorFlowConfig {

    @Bean
    public IntegrationFlow splitAggregateErrorFlow() {
        return IntegrationFlow.from("largeRequestInputChannel")
                // 拆分大请求为子消息并行处理
                .split()
                // 为子消息处理器添加异常捕获Advice
                .handle(subMessageProcessor(), config -> config.advice(processingResultAdvice()))
                // 聚合所有子消息的处理结果（等待全部子处理完成）
                .aggregate(aggregatorConfig -> aggregatorConfig
                        .outputProcessor(group -> group.getMessages()
                                .stream()
                                .map(msg -> (ProcessingResult<?>) msg.getPayload())
                                .collect(Collectors.toList()))
                        .sendPartialResultOnExpiry(false))
                // 统一处理聚合后的结果（分离成功/失败条目）
                .handle(aggregateResultHandler())
                .get();
    }

    // 模拟子消息业务处理逻辑（可能抛出异常）
    private MessageHandler subMessageProcessor() {
        return message -> {
            String payload = (String) message.getPayload();
            if (payload.contains("error")) {
                throw new RuntimeException("Processing failed for: " + payload);
            }
            // 成功处理时返回成功结果
            MessageChannel replyChannel = (MessageChannel) message.getHeaders().getReplyChannel();
            replyChannel.send(MessageBuilder.withPayload(ProcessingResult.success(payload)).build());
        };
    }

    // 配置ExpressionEvaluatingRequestHandlerAdvice，捕获异常并返回结构化结果
    @Bean
    public ExpressionEvaluatingRequestHandlerAdvice processingResultAdvice() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        // 成功时返回ProcessingResult.success(原始Payload)
        advice.setOnSuccessExpressionString("T(com.yourpackage.ProcessingResult).success(#root.payload)");
        // 失败时返回ProcessingResult.failure(原始Payload, 异常对象)
        advice.setOnFailureExpressionString("T(com.yourpackage.ProcessingResult).failure(#root.payload, #exception)");
        // 标记异常已处理，避免向上抛出中断流
        advice.setReturnFailureExpressionResult(true);
        return advice;
    }

    // 聚合结果统一处理逻辑
    private MessageHandler aggregateResultHandler() {
        return message -> {
            List<ProcessingResult<?>> allResults = (List<ProcessingResult<?>>) message.getPayload();
            
            // 分离成功和失败的结果
            List<Object> successfulPayloads = allResults.stream()
                    .filter(ProcessingResult::isSuccess)
                    .map(ProcessingResult::getPayload)
                    .collect(Collectors.toList());
            
            List<Map.Entry<Object, Throwable>> failedItems = allResults.stream()
                    .filter(result -> !result.isSuccess())
                    .map(result -> new AbstractMap.SimpleEntry<>(result.getPayload(), result.getException()))
                    .collect(Collectors.toList());

            // 处理成功结果逻辑
            System.out.println("Successfully processed payloads: " + successfulPayloads);

            // 统一处理异常逻辑（可根据原始Payload执行补偿、日志记录等操作）
            if (!failedItems.isEmpty()) {
                failedItems.forEach(item -> {
                    System.out.printf("Failed to process payload [%s], error: %s%n",
                            item.getKey(), item.getValue().getMessage());
                    // 此处添加基于Payload的后续业务逻辑，比如重试、补偿等
                });
            }
        };
    }
}

关键细节说明 #

• 异常捕获逻辑：ExpressionEvaluatingRequestHandlerAdvice通过onFailureExpressionString直接将原始Payload和异常封装为ProcessingResult.failure()，确保异常不中断并行流，同时保留原始业务数据。
• 聚合器配置：sendPartialResultOnExpiry(false)确保必须等待所有子消息处理完成后才执行聚合，满足"所有子消息处理完成后统一处理异常"的需求。
• 结果分离处理：聚合后通过ProcessingResult的isSuccess()方法快速区分成功/失败条目，分别执行业务逻辑，同时可直接通过getPayload()和getException()获取独立对象。

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