java - Spring 集成: Exception Handling on Publish/Subscribe Channels with Ordered Subscribers

Question

给定一个配置有任务执行器的发布-订阅 channel ，如果抛出异常，是否可以中断其有序订阅者的调用？

例如，在此示例中，“已工作”消息仍然由序列中的第二个服务激活器发送。我希望这种事不要发生。

    <int:publish-subscribe-channel id="input" task-executor="taskExecutor" />

    <int:service-activator
        input-channel="input"
        order="1" 
        expression="1 / 0" 
        output-channel="nullChannel"
    />
    <int:service-activator
        input-channel="input"
        order="2" 
        expression="'worked'" 
        output-channel="output"
    />

    <int:channel id="output">
        <int:queue />
    </int:channel>

现实世界的用例稍微复杂一些，因为一个 File 被发布给两个订阅者，一个用于解析其内容，另一个用于存档到 S3。应用一个序列，以便在两个任务完成后可以聚合和删除它们。

S3 上传是通过不返回回复的出站 channel 适配器实现的。因此，有一个发布-订阅 channel 首先调用 S3 适配器(使用 order 属性)，然后将 File 发送到桥接器，桥接器将其放置在收集聚合消息的 channel 。

如果S3上传失败，我不希望文件被删除，所以不能调用上游 channel 的第二个订阅者。

Answer 1

当您添加任务执行器时，“order”属性实际上没有任何意义，因为两个任务都是异步执行的；中断“第二个”线程可能为时已晚。要回答您的一般性问题，不，一个线程的失败不会中断其他线程。

如果我正确理解您的用例，您应该能够使用新的 2.2 <request-handler-advice-chain/> 执行您想要的操作功能，您可以在其中添加 ExpressionEvaluatingRequestHandlerAdvice根据成功或失败采取不同的行动。本博客对此进行了讨论... http://blog.springsource.org/2012/10/09/spring-integration-2-2-retry-and-more/并且，特别是它引用的示例代码(对于 ftp 适配器)...

<int-ftp:outbound-channel-adapter
    channel="inputChannel"
    session-factory="mockSessionFactory"
    remote-directory="foo">
    <int-ftp:request-handler-advice-chain>
        <bean class="org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice">
            <property name="onSuccessExpression" value="payload.delete()" />
            <property name="successChannel" ref="afterSuccessDeleteChannel" />
            <property name="onFailureExpression" value="payload.renameTo(payload.absolutePath + '.failed.to.send')" />
            <property name="failureChannel" ref="afterFailRenameChannel" />
        </bean>
    </int-ftp:request-handler-advice-chain>
</int-ftp:outbound-channel-adapter>