如何使用Spring集成DSL处理拆分的消息集合，并将故障传递给Spring云数据流

Question

我在使用DSL获取Spring集成流时遇到了一些问题，DSL有一个拆分器，可以在Spring云数据流上下文中正确地满足我的需要。从本质上讲，我的微服务是一个处理器，它试图做到以下几点：

1. 一些第三方报告被检索并合并在一起。这将生成一组事件。（注意：我的微服务不是一个源，因为它需要由一个可配置的时间表驱动，所以我有一个定时触发器，将命令消息传送到此微服务。）
2. 事件集合被拆分，因此可以验证每个事件。通过验证的事件很好，并继续进行。验证失败的事件在数据流上下文中必须是死信。因此，这里没有失败或成功的一批事件的概念

我试图克服的问题是：

• 对验证失败的事件抛出MessaginException不会导致进一步的事件处理
• 解决这种过早的停止会导致数据流中没有死信，因为解决方法涉及在执行器中抛出异常，当然这些异常不会发生

验证失败的事件用hasError消息头标记，路由器使用该消息头将其发送给引发MessaginException的ServiceActivator。

经过一些调查和实验，我有缺陷的实验如下所示：

IntegrationFlows.from(Processor.INPUT).

    // stuff ommitted for brevity
    handle(new MyEventPublisher(.........)).

    // List of events produced, split them
    split().

    // validate each event
    transform(new MyEventValidator()).

    // attempt to circumvent premature stoppage
    channel(MessageChannels.executor(Executors.newCachedThreadPool())).

    // route events based on validation result
    <String>route("headers[hasError] != null && headers[hasError] == 'true'",
    spec -> {
                spec.resolutionRequired(false);
                spec.defaultOutputChannel(Processor.OUTPUT);

                // A failed event routes to a service that throws a MessagingException
                spec.subFlowMapping("true", sf -> sf.<String>handle(new ExceptionThrowingService()));

                // Otherwise events flow onwards
                spec.channelMapping("false", Processor.OUTPUT);
            }).
get();

如果没有通道步骤和缓存的线程池，当遇到第一个事件验证失败时，处理会停止，但失败的事件是死信的，任何成功的事件都会通过。

使用线程池，所有事件都被处理。但是，在Dataflow上下文中没有事件是死记硬背的，因为异常是在执行程序线程中引发的，但成功的事件确实会引发Dataflow流。

我是否能够使用拆分器，处理整个输入，并将消息异常传递给数据流运行时？

Answer 1

死锁与入站消息有关。确实，每当下游发生任何异常时，接收传入消息的容器都会将其（或基于它的某些内容）发送到DLQ。这已经是您通过拆分器生成许多消息的自定义逻辑，但从代理的角度来看，这仍然是单个传入消息的问题。

为了克服这个问题，您应该考虑在每个拆分器项目验证时手动发送到DLQ。为此，您可以查看ExpressionEvalatingRequest estHandlerAdures：http://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#expression-advice.并将异常发送到某些通道进行分析。已经在那里，您可以发送到DLQ或其他东西。

在Spring Cloud Stream和Data Flow透视图中，您可以为目标添加一个更多的绑定配置：http://docs.spring.io/spring-cloud-stream/docs/Chelsea.SR2/reference/htmlsingle/index.html#__code_input_code_and_code_output_code