Spring集成–强大的拆分器聚合器

坚固是什么意思？

在本文的上下文中，健壮性是指在不立即返回到调用者的情况下管理流中的异常条件的能力。在某些处理方案中， n个 m个回答足以做出结论。通常具有这些趋势的示例处理场景是：

财务，保险和预订系统的报价。
扇出出版系统。

为什么需要鲁棒的分离器聚合器设计？

首先，可能需要对典型的Splitter Aggregator模式进行介绍。拆分器是一种EIP模式，它描述了一种机制，用于将复合消息分解为多个部分，以便可以分别处理它们。路由器是一种EIP模式，用于将消息路由到各个通道中-将它们瞄准特定的消息传递端点。聚合器是一种EIP模式，用于整理和存储属于某个组的一组消息，并在该组完成后释放它们。

这三个EIP构造共同构成了一种强大的机制，可将处理划分为不同的工作单元。 Spring Integration（SI）使用与EIP相同的模式术语，因此该方法的读者将非常熟悉Spring Integration Framework的构造。 SI框架允许对所有这三个结构进行重大定制，此外，就像在任何其他多线程配置中一样，只需使用异步通道即可使这些工作单元并行执行。

与SI Splitter Aggregator设计一起使用时，一个有趣的挑战是构建适当健壮的流，这些流可以在许多调用场景中可预测地运行。一个简单的拆分器聚合器设计可以在许多情况下使用，并且无需大量定制SI构造即可运行。但是，某些服务要求需要更强大的处理策略，因此需要更复杂的配置。以下各节描述并显示了Simple Splitter Aggregator设计的实际外观，设计必须能够处理的处理类型，然后为更健壮的处理提出建议的解决方案。

简单的拆分器聚合器设计

以下Splitter Aggregator设计展示了一个简单的流程，该流程将文档请求消息接收到消息传递网关中，将消息分为两个处理路由，然后聚合响应。请注意，该图是从OmniGraffle中的EIP构造构建的，而不是从STS内部的“集成图”视图。为了简洁起见，图中没有显示通道。

SI的详细构造：

消息传递网关 –有三个消息传递网关。有多种配置可用于网关规范，但可以显着地返回业务对象，异常和空值（超时后）。最左边的网关是我们为其定义流程的服务网关。路由器和聚合器之间的其他两个网关是外部系统，它们将提供对我们流程产生的业务问题的响应。

拆分器 –存在一个拆分器，它负责使用文档消息并生成消息集合以进行后续处理。通常，自定义拆分器的Java签名指定单个对象参数和用于返回的集合。

收件人列表路由器 –存在一个路由器，可以使用任何适当的路由器，选择与您的要求非常匹配的路由器–您可以轻松地按表达式或有效负载类型进行路由。路由器的主要目的是路由分离器提供的消息集合。这是一个非常典型的拆分器聚合器配置。

聚集器 –单个构造，负责将消息集中在一起收集，以便可以对网关响应进行进一步处理。尽管可以使用属性和Bean定义来配置Aggregator，以提供替代的分组和发布策略，但大多数情况下，默认的聚合策略就足够了。

拆分器聚合器操作的有趣方面

网关 –入站网关（最左边的一个）上可能有也可能没有定义错误处理bean引用。如果这样做，那么该bean将有机会处理该网关右侧的流中引发的异常。否则，任何异常将直接抛出网关。
网关 –可以在每个网关上设置一个可选的默认答复超时 ，设置此值有很大的含义，请确保它们被很好地理解。超时将导致网关返回空值。如果上游网关也没有设置默认应答超时 ，则这是可能导致线程驻留的完全相同的条件。
分配器输入通道 –这可以是简单的直接通道，也可以是定义了调度程序的直接通道。如果通道指定了调度程序，则此点的下游流将是异步的多线程的。这也改变了上游网关的语义，因为它通常意味着原本不重要的default-reply-timeout变为活动状态。
拆分器 –拆分器必须返回单个对象。拆分器返回的单个对象是一个集合java.util.List。 SI框架将采用该列表的每个成员，并将其馈入Splitter的输出通道 –与本示例一样，通常直接进入路由器。 “拆分器列表”返回的合同与在Java中的用法相同-它可能包含零个，一个或多个元素。如果拆分器返回一个空列表，则路由器不太可能要做任何工作，因此流调用将完成。但是，如果列表包含一项，则SI框架将从列表中提取该项目并将其推入路由器，如果成功路由，则流程将继续。
路由器 –在此示例中，路由器将仅将消息路由到两个网关之一。
网关 –在拆分器和聚合器之间使用的两个网关很有趣。在此示例中，我使用通用网关EIP模式表示消息子系统，但未明确定义-我们可以使用HTTP出站网关，另一个SI流或任何其他外部系统。当然，对于这些子系统中的每一个，许多响应都是可能的。取决于协议和外部系统，消息请求可能无法发送，响应未能到达，调用了长时间运行的进程，网络错误或超时或常规处理异常。
聚合器 –单个聚合器将等待大量响应，具体取决于拆分器创建的内容。在拆分器返回列表为空的情况下，将不会调用聚合器。如果拆分器返回列表只有一个条目，则聚合器将等待一个网关响应来完成该组。如果“拆分器”列表具有n个条目，则聚合器将等待n个条目来完成该组。可以在一组丰富的配置方面中注入自定义关联策略，发布策略和消息存储。

简单拆分器聚合器操作的有趣方面

确定这种类型的简单网关是否足以满足要求的主要决定因素是了解发生故障时发生的情况。如果您的SI流中发生任何异常导致流调用被放弃并且符合您的要求，则无需进一步阅读。但是，如果您需要在其中一个网关发生故障后继续进行处理，那么本文的其余部分可能会引起您的更多兴趣。

拆分器和聚合器之间生成的任何来源的异常都将导致聚合器丢弃空的或部分的组。异常将传播回最近的上游网关，以供自定义bean处理或由网关重新抛出。请注意，聚合器上的自定义释放策略很难使用，尤其是与超时一起使用，但在这种情况下无济于事，因为异常将在调用聚合器之前传播回最左边的网关。

也可以在最内部的网关上配置异常处理程序，可能会捕获到异常消息，但是如何将消息从定制异常处理程序路由到聚合器以完成组，将聚合器通道定义注入到定制异常处理程序中呢？这是一种较差的方法，可能涉及解包异常消息有效负载，将原始消息标头复制到新的SI消息中，然后添加原始有效负载–只有四或五行代码，但是很脏。

生成异常后，不能将异常消息（ 未经修改 ）路由到聚合器中以完成组。原始消息（包含有关组和组位置的相关ID和序列ID的消息）被掩埋在SI消息异常有效负载之内。

如果在异常生成之后需要继续处理，则必须清楚，为了继续处理，必须进行以下操作：

聚合组需要完成，
返回到壁橱上游网关之前，必须捕获和处理所有异常，
允许在聚合器中完成组的相关性和序列标识符被掩埋在异常消息有效负载之内，并且将需要对绑定到聚合器的消息进行提取和设置

更健壮的解决方案–消息传递网关适配器模式

处理网关的异常和空返回值自然会导致一种设计，该设计在消息传递网关周围实现包装器。这样就提供了一个很难建立的控制级别。

这种适配器技术允许在将消息传递网关注入到Service Activator并从中直接调用时，捕获并处理消息传递网关的所有返回。消息传递网关不再直接响应聚合器，而是响应在Service Activator名称空间定义中配置的自定义Java代码Spring bean。不出所料，不会发生异常的处理将继续正常进行。那些经历异常情况或消息网关发出意外响应或缺少响应的流需要以某种方式处理消息，例如可以完成绑定到聚合的消息组。如果Service Activator允许将异常传播到其支持bean之外，则该组将无法完成。同样的情况不仅适用于异常，而且不包含前提条件组相关ID和序列标头的任何返回对象-这就是应用适配的地方。

捕获和处理来自消息传递网关的异常消息或空响应，如以下示例代码所示：

import com.l8mdv.sample.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;public class AvsServiceImpl implements AvsService {private static final Logger logger= LoggerFactory.getLogger(AvsServiceImpl.class);public static final String MISSING_MANDATORY_ARG= "Mandatory argument is missing.";private AvsGateway avsGateway;public AvsServiceImpl(final AvsGateway avsGateway) {this.avsGateway = avsGateway;}public Message<AvsResponse> service(Message<AvsRequest> message) {Assert.notNull(message, MISSING_MANDATORY_ARG);Assert.notNull(message.getPayload(), MISSING_MANDATORY_ARG);MessageHeaders requestMessageHeaders = message.getHeaders();Message<AvsResponse> responseMessage = null;try {logger.debug("Entering AVS Gateway");responseMessage = avsGateway.send(message);if (responseMessage == null)responseMessage = buildNewResponse(requestMessageHeaders,AvsResponseType.NULL_RESULT);logger.debug("Exited AVS Gateway");return responseMessage;} catch (Exception e) {return buildNewResponse(responseMessage, requestMessageHeaders,AvsResponseType.EXCEPTION_RESULT, e);}}private Message<AvsResponse> buildNewResponse(MessageHeaders requestMessageHeaders,AvsResponseType avsResponseType) {Assert.notNull(requestMessageHeaders, MISSING_MANDATORY_ARG);Assert.notNull(avsResponseType, MISSING_MANDATORY_ARG);AvsResponse avsResponse = new AvsResponse();avsResponse.setError(avsResponseType);return MessageBuilder.withPayload(avsResponse).copyHeadersIfAbsent(requestMessageHeaders).build();}private Message<AvsResponse> buildNewResponse(Message<AvsResponse> responseMessage,MessageHeaders requestMessageHeaders,AvsResponseType avsResponseType,Exception e) {Assert.notNull(responseMessage, MISSING_MANDATORY_ARG);Assert.notNull(responseMessage.getPayload(), MISSING_MANDATORY_ARG);Assert.notNull(requestMessageHeaders, MISSING_MANDATORY_ARG);Assert.notNull(avsResponseType, MISSING_MANDATORY_ARG);Assert.notNull(e, MISSING_MANDATORY_ARG);AvsResponse avsResponse = new AvsResponse();avsResponse.setError(avsResponseType,responseMessage.getPayload(), e);return MessageBuilder.withPayload(avsResponse).copyHeadersIfAbsent(requestMessageHeaders).build();}
}

注意异常处理块的catch子句的最后一行。这行代码将相关性和序列标头复制到响应消息中，如果要允许聚合组完成，则这是强制性的，并且在出现异常后总是必需的，如此处所示。