Spring集成errorChannel，错误句柄

Question

场景可能是:我的期望可能是批量 10 个数据点，我想对 {failed 5, pass 5} 或 sth. 做出响应。

我的逻辑是将批次拆分为数据元素并进行验证。
成功的验证将发送到聚合器，
失败的验证将抛出错误并由错误 channel 接收。

接收者列表路由器将 errorChannel 作为 inputChannel 和 2 个过滤器连接到它，目的是过滤某种类型的错误以直接发送响应(与用户输入无关的接收 - 服务器错误等)，某种类型的客户端错误将转到聚合器以构建响应。

逻辑有问题吗？
我的问题是在聚合器之后使用服务激活器构建结果时，我不断收到“收到回复消息，但接收线程已经收到回复”。这个服务激活器连接到replyChannel，似乎有一些消息已经发送到这个 channel ？

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

有问题？顺便说一句，收件人列表路由器或其他类型的端点可以连接到 errorChannel 吗？或者它必须是服务激活器，就像我在所有在线示例中看到的那样？(但它们是非常简单的示例..)

示例 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

你的问题不清楚。请确保将来提供更具体的信息。一些配置和 StackTrace 或日志也很有用。

我猜你在你的流程开始时就有 <gateway>已配置 error-channel .这就是为什么您会收到 Reply message received but the receiving thread has already received a reply .

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

您不能依赖 errorChannel标题在那里，然后回来 reply最终因为errorChannel网关的情况下的 header 与 replyChannel 相同，它是 TemporaryReplyChannel - 一次性使用 channel ，仅适用于 receive()手术。

由于我们没有您提供的任何配置或代码，因此我们无法为您提供适当的帮助。

我想你需要一个中间流 gateway通过 service-activator :

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

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

或 <chain>与 <gateway>仅在那里执行验证并捕获错误并返回所需的失败回复。那个 service-activator将能够将它们全部发送到 <aggregator>之后。在这种情况下， <aggregator>可以正常回复网关。

编辑

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

因此，该名称对您的用例完全不利，因为 myErrorRouter将处理所有错误!

<int:recipient-list-router>如果通过 selector，则向其所有收件人发送消息或者如果没有 selector .

你不需要default-reply-channel在 <gateway>如果不是 publish-subscribe .您可以依靠 replyChannel标题，如果没有 output-channel 则有效defined .

我在说什么 <service-activator>和 <gateway>在您的 <transformer> 面前非常直截了当之后 <splitter> :

<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>

所以， splitter将项目发送到 service-activator .那个人继续向 gateway 发送消息附近 transformer定制 error-channel正好为一项。 transformer回复 replyChannel与之前的完全一致 gateway .如果抛出异常，则由 validationErrorChannel 处理。过程。应该回复一些补偿信息。该消息冒泡到 service-activator .最后 service-activator将结果发送到 aggregateChannel .它是 service-activator 的黑匣子验证是否良好。

希望那些对你有帮助。

编辑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 上进行错误处理。 .