spring-integration：在 Aggregator 之后发起事务答案

【问题标题】：spring-integraton: Initiate a transaction after an Aggregatorspring-integration：在 Aggregator 之后发起事务
【发布时间】：2021-01-22 02:47:45
【问题描述】：

我必须以事务方式处理一些按名称聚合它们的文件组：只是说我有这样命名的输入文件：A1, A2, B1, B2, C1, C2, ...

我需要像这样处理它们在列表中分组：

[A1, A2], [B1, B2], [C1, C2]...

我用一个 FTP inboundAdapter、一个聚合器和我的自定义处理器创建了一个通用的事务链，它运行良好，除了一个细节。

当列表处理失败时，回滚 SpEL 仅处理触发异常的单个文件，而我想处理它所属的文件列表的异常。

我基本上（认为）我需要的是这样的：

+--+  +--+  +--+  +--+  +--+  +--+                       +------+  +------+  +------+
|A1|  |A2|  |B1|  |B2|  |C1|  |C2|     |aggregator| ->   |A1, A2|  |B1, B2|  |C1, C2|
+--+  +--+  +--+  +--+  +--+  +--+                       +------+  +------+  +------+
                            
                                                       \-----------------------------/
                                                              transaction's scope

如果我只在聚合器之后启动事务，SpEL 表达式将只看到列表消息而不是单个文件，对吗？

我不知道该怎么做，是在不使用链中间的轮询器的情况下启动交易，这会引入延迟。

有什么提示吗？

谢谢！

编辑：我已经解决了在处理句柄中添加一个建议，但我不能 100% 确定它是“正确”的解决方案

【问题讨论】：

是的；这是正确的解决方案；我建议您添加一个答案（并接受它）显示您修改后的代码。

标签： spring spring-integration

【解决方案1】：

按照 Gary 的建议，这是我采用的解决方案：

return IntegrationFlows
  .from(Ftp.inboundAdapter(ftpSessionFactory)
     // my FTP inbound adapter
  )
  .aggregate(a ->
     //  my aggregator strategy     
  )
  .handle(myAggregatedMessageProcessor(), c -> c.advice(myProcessorAdvice()))
  .get();

相关部分为c -> c.advice(processErrorAdvice())

我还配置了这样的建议：

  @Bean
  Advice myProcessorAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setSuccessChannelName("successChannel");
        advice.setFailureChannelName("failedChannel");
        
        // this catches the Exceptions
        advice.setTrapException(true);
        return advice;
    }

【讨论】：