Spring集成错误通道，错误处理

Question

场景可能是：我的期望可能是批量10个数据点，我想对{failed 5，pass 5}或其他什么给出响应。

我的逻辑是将批处理拆分为数据元素并进行验证
成功的验证将发送给aggreagtor，
失败的验证将抛出错误并通过错误通道拾取。

收件人列表路由器将错误通道作为输入通道，并连接2个过滤器，目的是过滤某些类型的错误直接发送响应（与用户输入无关的信息-服务器错误等），某些类型的客户端错误将转到聚合器构建响应。

逻辑有问题吗？我的问题是我一直得到"回复消息收到，但接收线程已经收到回复"当建立使用服务激活器后聚合的结果。

在“错误过滤器”连接到replyChannel后，我只检查了此服务激活器和服务器错误分支的集成工作流程（但从未调用句柄）

有什么不对吗？顺便说一句，收件人列表路由器或其他类型的endpoint可以连接到错误通道吗？或者它必须是服务激活器，就像我在网上看到的所有例子一样？但他们真的很简单的例子...（

示例XML

<int:gateway id="myGateway" service-interface="someGateway" default-request-channel="splitChannel" error-channel="errorChannel"  default-reply-channel="replyChannel" async-executor="MyThreadPoolTaskExecutor"/>

<int:splitter input-channel="splitChannel" output-channel="transformChannel" method="split">
    <bean class="Splitter" />
</int:splitter>

<int:transformer id="transformer" input-channel="transformChannel" method="transform" output-channel="aggregateChannel">
    <bean class="Transformer"/>  // this may throw the validation error (filter_ErrorType_1), if it cannot transform
</int:transformer>

<int:aggregator id="aggregator"
    input-channel="aggregateChannel"
    output-channel="createAnswerChannel"
    method="aggregate">
    <bean class="MyAggregator" />
</int:aggregator>

<int:recipient-list-router id="myErrorRouter" input-channel="errorChannel">
    <int:recipient channel="filter_ErrorType_1"/> 
    <int:recipient channel="filter_ErrorType_2"/> 
    <int:recipient channel="filter_ErrorType_3"/> 
</int:recipient-list-router>

<int:filter input-channel="filter_ErrorType_1" output-channel="aggregateChannel" method="accept"></int:filter>
<int:filter input-channel="filter_ErrorType_2" output-channel="createErrorAnswerChannel" method="accept"></int:filter>
<int:filter input-channel="filter_ErrorType_3" output-channel="createErrorAnswerChannel" method="accept"></int:filter>

<int:service-activator input-channel='createErrorAnswerChannel' output-channel="replyChannel" method='buildError'>
    <bean class="AnswerBuilder"/>
</int:service-activator>

<int:service-activator input-channel='createAnswerChannel' output-channel="replyChannel" method='build'>
    <bean class="AnswerBuilder"/>
</int:service-activator>

跟进：

<int:gateway id="myGateway" service-interface="someGateway" default-request-channel="splitChannel" error-channel="errorChannel"  default-reply-channel="replyChannel" async-executor="MyThreadPoolTaskExecutor"/>

<int:splitter input-channel="splitChannel" output-channel="transformChannel" method="split">
    <bean class="Splitter" />
</int:splitter>

<int:transformer id="transformer1" input-channel="toTransformer1" method="transform" output-channel="toTransformer2">
    <bean class="Transformer1"/>  // this may throw the validation error (filter_ErrorType_1), if it cannot transform
</int:transformer>

<int:transformer id="transformer2" input-channel="toTransformer2" method="transform" output-channel="toTransformer3">
    <bean class="Transformer2"/>  // this may throw the validation error (filter_ErrorType_2), if it cannot transform
</int:transformer>

<int:transformer id="transformer3" input-channel="toTransformer3" method="transform" output-channel="aggregateChannel">
    <bean class="Transformer3"/>  // this may throw the validation error (filter_ErrorType_3), if it cannot transform
</int:transformer>

???
// seems like you are proposing to have one gateway for each endpoint that may throw error.
// but in this case, take transfomer 1 for example, I cannot output the gateway directly to aggregate channel since for valid data it has to go to transformer 2
// but the failed message throwed by the error handler cannot pass transformer 2 because of afterall this is a error message not a valid data for transformer 2
// <int:service-activator input-channel="toTransformer1" output-channel="toTransformer2" ref="gateway1"/>

// <int:gateway id="gateway1" default-request-channel="toTransformer1" error-channel="errorChannel1"/>

// <int:transformer id="transformer" input-channel="toTransformer1" method="transform">
//     <bean class="Transformer"/>  // this may throw the validation error (filter_ErrorType_1), if it cannot transform
// </int:transformer>

// how to deal with this problem?




<int:service-activator input-channel='createErrorAnswerChannel' output-channel="replyChannel" method='buildError'>
    <bean class="AnswerBuilder"/>
</int:service-activator>

<int:service-activator input-channel='createAnswerChannel' output-channel="replyChannel" method='build'>
    <bean class="AnswerBuilder"/>
</int:service-activator>

我实际上有复杂的逻辑，但我用transformer 123来表示。

Answer 1

你的问题不清楚。请务必在将来提供更具体的信息。一些配置和堆栈跟踪或日志也很有用。

我猜你已经在流的开始

但我不能确定，因为你的问题中没有这些词。正当

您不能依赖那里的errorChannel头，最终返回reply，因为网关情况下的errorChannel头与replyChannel相同，它是TemporaryReplyChannel-一次性使用通道，仅用于receive（）操作。

由于我们没有您的任何配置或代码，我们无法正确地帮助您。

我想你需要一个中间流网关通过服务激活器：

<service-activator id="gatewayTestService" input-channel="inputChannel" 
            output-channel="outputChannel" ref="gateway"/>

<gateway id="gateway" default-request-channel="requestChannel" error-channel="myErrorChannel"/>

或

编辑

1. errorChannel是默认全局bean的名称，用于捕获来自任何集成位置的任何错误消息。在http://docs.spring.io/spring-integration/reference/html/configuration.html#namespace-errorhandler.中查看更多信息

因此，这个名称对您的用例来说是完全不好的，因为myErrorOuter将处理所有错误！

在

我说的是

<int:service-activator input-channel="transformChannel" 
            output-channel="aggregateChannel" ref="gateway"/>

<int:gateway id="gateway" default-request-channel="transformChannel" error-channel="validationErrorChannel"/>

<int:transformer id="transformer" input-channel="transformChannel" method="transform">
    <bean class="Transformer"/>  // this may throw the validation error (filter_ErrorType_1), if it cannot transform
</int:transformer>

因此，拆分器将项目发送到服务激活器。通过自定义的错误通道将消息发送到网关周围的变压器。transformer对replyChannel的应答与前面的网关的应答完全一致。如果它抛出一些异常，则由validationErrorChannel进程处理。它应该回复一些补偿信息。该消息被冒泡发送到服务激活器。最后，service activator将结果发送到aggregateChannel。如果验证良好与否，服务激活器的黑匣子。

希望这有点帮助。

编辑2

我很失望，您没有接受我在代码中的建议，但这就是我的看法：

<int:gateway id="myGateway" service-interface="someGateway" default-request-channel="splitChannel"
             async-executor="MyThreadPoolTaskExecutor" />

<int:splitter input-channel="splitChannel" output-channel="transformChannel" method="split">
    <bean class="Splitter" />
</int:splitter>

<int:service-activator input-channel="validateChannel" output-channel="aggregateChannel"
      ref="validateGateway"/>

<gateway id="validateGateway" default-request-channel="toTransformer1" error-channel="myErrorChannel"/>

<chain input-channel="toTransformer1">
    <int:transformer method="transform">
        <bean class="Transformer1" />
    </int:transformer>
    <int:transformer method="transform">
        <bean class="Transformer2" />
    </int:transformer>
    <int:transformer method="transform">
        <bean class="Transformer3" />
    </int:transformer>
</chain>


<int:service-activator input-channel="myErrorChannel" method="buildError">
    <bean class="AnswerBuilder" />
</int:service-activator>

<int:aggregator id="aggregator"
                input-channel="aggregateChannel"
                output-channel="createAnswerChannel"
                method="aggregate">
    <bean class="MyAggregator" />
</int:aggregator>

<int:service-activator input-channel='createAnswerChannel' method='build'>
    <bean class="AnswerBuilder" />
</int:service-activator>

注意，我是如何链接变形金刚的。因此，您的所有变压器都有一个网关，其中任何一个变压器上的任何错误都将被抛出到myErrorChannel上的网关进行错误处理。