消息处理器链

MessageHandlerChainMessageHandler 的一个实现,它可以配置为单个消息端点,同时实际委托给其他处理器的链,例如过滤器、转换器、分发器等。当多个处理器需要以固定的线性顺序连接时,这可以带来更简单的配置。例如,在其他组件之前提供转换器是相当常见的。类似地,当你在链中的其他组件之前提供过滤器时,你实际上创建了一个选择性消费者(selective consumer)。在任何一种情况下,链都只需要一个 input-channel 和一个 output-channel,从而消除了为每个独立组件定义通道的需要。

MessageHandlerChain 主要设计用于 XML 配置。对于 Java DSL,IntegrationFlow 定义可以被视为链组件,但这与本章下面描述的概念和原则无关。有关更多信息,请参阅Java DSL
Spring Integration 的 Filter 提供了一个布尔属性:throwExceptionOnRejection。当你在同一个点对点通道上提供多个具有不同接受条件的 selective consumer 时,你应该将此值设置为 'true'(默认值为 false),以便调度器知道消息已被拒绝,因此尝试将消息传递给其他订阅者。如果未抛出异常,则调度器会认为消息已成功传递,即使过滤器已丢弃消息以防止进一步处理。如果确实想“丢弃”消息,过滤器的 'discard-channel' 可能很有用,因为它确实为你提供了对丢弃的消息执行某些操作的机会(例如将其发送到 JMS 队列或写入日志)。

处理器链简化了配置,同时在内部保持了组件之间相同程度的松散耦合,并且如果在某个时刻需要非线性排列,修改配置也非常简单。

在内部,链被扩展为所列端点的线性设置,由匿名通道分隔。链内不考虑回复通道(reply channel)头。只有在最后一个处理器被调用后,结果消息才会被转发到回复通道(reply channel)或链的输出通道(output channel)。由于这种设置,除最后一个处理器外,所有处理器都必须实现 MessageProducer 接口(该接口提供了 'setOutputChannel()' 方法)。如果在 MessageHandlerChain 上设置了 outputChannel,则最后一个处理器只需要一个 output channel。

与其他端点一样,output-channel 是可选的。如果在链的末尾有回复消息,output-channel 具有优先权。但是,如果它不可用,链处理器会检查入站消息上的回复通道头(reply channel header)作为备用。

在大多数情况下,你无需自己实现 MessageHandler。下一节重点介绍 chain 元素的命名空间支持。大多数 Spring Integration 端点,例如 service activators 和 transformers,都适合在 MessageHandlerChain 中使用。

配置链

<chain> 元素提供了 input-channel 属性。如果链中的最后一个元素能够产生回复消息(可选),它也支持 output-channel 属性。其子元素可以是过滤器、转换器、分发器和服务激活器。最后一个元素也可以是路由器或出站通道适配器。以下示例展示了链的定义

<int:chain input-channel="input" output-channel="output">
    <int:filter ref="someSelector" throw-exception-on-rejection="true"/>
    <int:header-enricher>
        <int:header name="thing1" value="thing2"/>
    </int:header-enricher>
    <int:service-activator ref="someService" method="someMethod"/>
</int:chain>

前面示例中使用的 <header-enricher> 元素在消息上设置了一个名为 thing1、值为 thing2 的消息头。header enricher 是 Transformer 的一种特殊形式,它仅修改 header 值。你可以通过实现一个 MessageHandler 来完成这些 header 修改并将其配置为一个 bean,从而获得相同的结果,但 header-enricher 是一种更简单的选择。

可以将 <chain> 配置为消息流的最后一个“封闭式”消费者。为此,可以在 <chain> 的末尾放置一些 <outbound-channel-adapter>,如下例所示

<int:chain input-channel="input">
    <int-xml:marshalling-transformer marshaller="marshaller" result-type="StringResult" />
    <int:service-activator ref="someService" method="someMethod"/>
    <int:header-enricher>
        <int:header name="thing1" value="thing2"/>
    </int:header-enricher>
    <int:logging-channel-adapter level="INFO" log-full-message="true"/>
</int:chain>
不允许的属性和元素

链中使用的组件不允许指定某些属性,例如 orderinput-channelpoller 子元素也是如此。

对于 Spring Integration 核心组件,XML 模式本身强制执行部分约束。然而,对于非核心组件或你自己的自定义组件,这些约束是由 XML 命名空间解析器强制执行的,而不是由 XML 模式强制执行的。

这些 XML 命名空间解析器约束是在 Spring Integration 2.2 中添加的。如果你尝试使用不允许的属性和元素,XML 命名空间解析器会抛出 BeanDefinitionParsingException

使用 'id' 属性

从 Spring Integration 3.0 开始,如果 chain 元素被赋予 id 属性,则该元素的 bean 名称是链的 id 与元素自身 id 的组合。没有 id 属性的元素不会注册为 bean,但每个元素都会获得一个包含链 idcomponentName。考虑以下示例

<int:chain id="somethingChain" input-channel="input">
    <int:service-activator id="somethingService" ref="someService" method="someMethod"/>
    <int:object-to-json-transformer/>
</int:chain>

在前述示例中

  • <chain> 根元素的 id 为 'somethingChain'。因此,AbstractEndpoint 实现(PollingConsumerEventDrivenConsumer,取决于 input-channel 类型)的 bean 将此值作为其 bean 名称。

  • MessageHandlerChain bean 获取一个 bean 别名 ('somethingChain.handler'),这允许从 BeanFactory 直接访问此 bean。

  • <service-activator> 不是一个完全成熟的消息端点(它不是 PollingConsumerEventDrivenConsumer)。它是 <chain> 中的一个 MessageHandler。在这种情况下,在 BeanFactory 中注册的 bean 名称是 'somethingChain$child.somethingService.handler'。

  • ServiceActivatingHandlercomponentName 与上述值相同,但不包含 '.handler' 后缀。它变为 'somethingChain$child.somethingService'。

  • 最后一个 <chain> 子组件,<object-to-json-transformer>,没有 id 属性。它的 componentName 基于其在 <chain> 中的位置。在本例中,它是 'somethingChain$child#1'。(名称的最后一个元素是其在链中的顺序,从 '#0' 开始)。请注意,此转换器未在应用程序上下文中注册为 bean,因此没有 beanName。但是,它的 componentName 有一个值,对于日志记录和其他目的非常有用。

<chain> 元素的 id 属性使其有资格进行JMX 导出,并且可以在消息历史中追踪它们。你可以通过使用适当的 bean 名称从 BeanFactory 访问它们,如前所述。

<chain> 元素上提供显式的 id 属性有助于简化日志中子组件的识别,并允许从 BeanFactory 等地方访问它们。

从链内部调用链

有时,你需要从一个链的内部嵌套调用另一个链,然后返回并在原始链中继续执行。为此,你可以使用消息网关,通过包含一个 <gateway> 元素来实现,如下例所示

<int:chain id="main-chain" input-channel="in" output-channel="out">
    <int:header-enricher>
      <int:header name="name" value="Many" />
    </int:header-enricher>
    <int:service-activator>
      <bean class="org.foo.SampleService" />
    </int:service-activator>
    <int:gateway request-channel="inputA"/>
</int:chain>

<int:chain id="nested-chain-a" input-channel="inputA">
    <int:header-enricher>
        <int:header name="name" value="Moe" />
    </int:header-enricher>
    <int:gateway request-channel="inputB"/>
    <int:service-activator>
        <bean class="org.foo.SampleService" />
    </int:service-activator>
</int:chain>

<int:chain id="nested-chain-b" input-channel="inputB">
    <int:header-enricher>
        <int:header name="name" value="Jack" />
    </int:header-enricher>
    <int:service-activator>
        <bean class="org.foo.SampleService" />
    </int:service-activator>
</int:chain>

在前面的示例中,main-chain 处理结束时由配置在那里的 'gateway' 元素调用 nested-chain-a。在 nested-chain-a 中,进行 header enrichment 后会调用 nested-chain-b。然后流程返回并在 nested-chain-b 中完成执行。最后,流程返回到 main-chain。当在链中定义嵌套版本的 <gateway> 元素时,它不需要 service-interface 属性。相反,它获取消息的当前状态并将其放置在 request-channel 属性定义的通道上。当由该网关启动的下游流完成时,一个 Message 被返回到网关并在当前链中继续其流程。