幂等接收器企业集成模式

从 4.1 版本开始，Spring Integration 提供了幂等接收器企业集成模式的实现。它是一个功能模式，整个幂等逻辑应该在应用程序中实现。但是，为了简化决策，提供了 IdempotentReceiverInterceptor 组件。这是一个 AOP Advice，它应用于 MessageHandler.handleMessage() 方法，并且可以根据其配置 过滤 请求消息或将其标记为 重复。

以前，您可以通过在 <filter/> 中使用自定义的 MessageSelector 来实现这种模式（参见过滤器），例如。但是，由于这种模式实际上定义了端点的行为，而不是端点本身，因此幂等接收器实现不提供端点组件。相反，它应用于应用程序中声明的端点。

IdempotentReceiverInterceptor 的逻辑基于提供的 MessageSelector，如果消息未被该选择器接受，则会用 duplicateMessage 标头设置为 true 来丰富它。目标 MessageHandler（或下游流程）可以查询此标头以实现正确的幂等逻辑。如果 IdempotentReceiverInterceptor 配置了 discardChannel 或 throwExceptionOnRejection = true，则不会将重复消息发送到目标 MessageHandler.handleMessage()。相反，它会被丢弃。如果您想丢弃（对重复消息不做任何处理），则应将 discardChannel 配置为 NullChannel，例如默认的 nullChannel bean。

为了在消息之间维护状态并提供比较消息以实现幂等性的能力，我们提供了 MetadataStoreSelector。它接受一个 MessageProcessor 实现（它根据 Message 创建一个查找键）和一个可选的 ConcurrentMetadataStore (元数据存储)。有关更多信息，请参见 MetadataStoreSelector Javadoc。您还可以使用额外的 MessageProcessor 自定义 ConcurrentMetadataStore 的 value。默认情况下，MetadataStoreSelector 使用 timestamp 消息标头。

通常，如果键不存在现有值，则选择器会选择接受消息。在某些情况下，比较键的当前值和新值以确定是否应接受消息非常有用。从 5.3 版本开始，提供了 compareValues 属性，它引用 BiPredicate<String, String>；第一个参数是旧值；返回 true 以接受消息并将旧值替换为 MetadataStore 中的新值。这对于减少键的数量很有用；例如，在处理文件中的行时，您可以在键中存储文件名，在值中存储当前行号。然后，在重新启动后，您可以跳过已经处理过的行。有关示例，请参见幂等下游处理拆分文件。

为了方便起见，MetadataStoreSelector 选项可以直接在 <idempotent-receiver> 组件上配置。以下清单显示了所有可能的属性

<idempotent-receiver
        id=""  (1)
        endpoint=""  (2)
        selector=""  (3)
        discard-channel=""  (4)
        metadata-store=""  (5)
        key-strategy=""  (6)
        key-expression=""  (7)
        value-strategy=""  (8)
        value-expression=""  (9)
        compare-values="" (10)
        throw-exception-on-rejection="" />  (11)

1	`IdempotentReceiverInterceptor` bean 的 ID。可选。
2	应用此拦截器的消费者端点名称或模式。用逗号 (`,`) 分隔名称（模式），例如 `endpoint="aaa, bbb, ccc, ddd, eee*fff"`。与这些模式匹配的端点 Bean 名称将用于检索目标端点的 `MessageHandler` Bean（使用其 `.handler` 后缀），并将 `IdempotentReceiverInterceptor` 应用于这些 Bean。必需。
3	一个 `MessageSelector` Bean 引用。与 `metadata-store` 和 `key-strategy (key-expression)` 互斥。当未提供 `selector` 时，需要 `key-strategy` 或 `key-strategy-expression` 之一。
4	标识当 `IdempotentReceiverInterceptor` 不接受消息时要发送消息的通道。省略时，重复消息将转发到带有 `duplicateMessage` 标头的处理程序。可选。
5	一个 `ConcurrentMetadataStore` 引用。由底层的 `MetadataStoreSelector` 使用。与 `selector` 互斥。可选。默认的 `MetadataStoreSelector` 使用一个内部 `SimpleMetadataStore`，该存储在应用程序执行之间不维护状态。
6	一个 `MessageProcessor` 引用。由底层的 `MetadataStoreSelector` 使用。从请求消息中评估 `idempotentKey`。与 `selector` 和 `key-expression` 互斥。当未提供 `selector` 时，需要 `key-strategy` 或 `key-strategy-expression` 之一。
7	一个 SpEL 表达式，用于填充 `ExpressionEvaluatingMessageProcessor`。由底层的 `MetadataStoreSelector` 使用。通过使用请求消息作为评估上下文根对象来评估 `idempotentKey`。与 `selector` 和 `key-strategy` 互斥。当未提供 `selector` 时，需要 `key-strategy` 或 `key-strategy-expression` 之一。
8	一个 `MessageProcessor` 引用。由底层的 `MetadataStoreSelector` 使用。从请求消息中评估 `idempotentKey` 的 `value`。与 `selector` 和 `value-expression` 互斥。默认情况下，'MetadataStoreSelector' 使用 'timestamp' 消息标头作为元数据 'value'。
9	用于填充`ExpressionEvaluatingMessageProcessor`的 SpEL 表达式。由底层的`MetadataStoreSelector`使用。通过使用请求消息作为评估上下文根对象，为`idempotentKey`评估`value`。与`selector`和`value-strategy`互斥。默认情况下，'MetadataStoreSelector' 使用 'timestamp' 消息头作为元数据 'value'。
10	对`BiPredicate<String, String>` bean 的引用，允许您选择性地通过比较键的旧值和新值来选择消息；默认情况下为`null`。
11	如果`IdempotentReceiverInterceptor`拒绝消息，是否抛出异常。默认为`false`。无论是否提供`discard-channel`，它都会应用。

对于 Java 配置，Spring Integration 提供方法级@IdempotentReceiver 注解。它用于标记具有消息传递注解（@ServiceActivator、@Router 等）的method，以指定哪些IdempotentReceiverInterceptor 对象应用于此端点。以下示例展示了如何使用@IdempotentReceiver 注解

@Bean
public IdempotentReceiverInterceptor idempotentReceiverInterceptor() {
   return new IdempotentReceiverInterceptor(new MetadataStoreSelector(m ->
                                                    m.getHeaders().get(INVOICE_NBR_HEADER)));
}

@Bean
@ServiceActivator(inputChannel = "input", outputChannel = "output")
@IdempotentReceiver("idempotentReceiverInterceptor")
public MessageHandler myService() {
    ....
}

当您使用 Java DSL 时，您可以将拦截器添加到端点的建议链中，如下面的示例所示

@Bean
public IntegrationFlow flow() {
    ...
        .handle("someBean", "someMethod",
            e -> e.advice(idempotentReceiverInterceptor()))
    ...
}

IdempotentReceiverInterceptor 仅针对MessageHandler.handleMessage(Message<?>) 方法设计。从 4.3.1 版本开始，它实现了HandleMessageAdvice，以AbstractHandleMessageAdvice 作为基类，以实现更好的分离。有关更多信息，请参阅处理消息建议。