幂等接收器企业集成模式
从 4.1 版本开始,Spring Integration 提供了 幂等接收器 企业集成模式的实现。这是一个功能模式,整个幂等逻辑应该在应用程序中实现。但是,为了简化决策过程,提供了 IdempotentReceiverInterceptor
组件。这是一个应用于 MessageHandler.handleMessage()
方法的 AOP Advice
,它可以根据其配置来 过滤
请求消息或将其标记为 重复
消息。
以前,您可以通过在 <filter/>
中使用自定义 MessageSelector
来实现此模式(例如,参见 过滤器)。但是,由于此模式实际上定义了端点的行为,而不是端点本身,因此幂等接收器实现不提供端点组件。相反,它应用于应用程序中声明的端点。
IdempotentReceiverInterceptor
的逻辑基于提供的 MessageSelector
,如果消息未被该选择器接受,则会使用设置为 true
的 duplicateMessage
标头对其进行丰富。目标 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 | 用于填充 ExpressionEvaluatingMessageProcessor 的 SpEL 表达式。由底层的 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
等)的方法,以指定哪些 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 时,您可以将拦截器添加到端点的 advice 链中,如下例所示
@Bean
public IntegrationFlow flow() {
...
.handle("someBean", "someMethod",
e -> e.advice(idempotentReceiverInterceptor()))
...
}
IdempotentReceiverInterceptor 仅设计用于 MessageHandler.handleMessage(Message<?>) 方法。从 4.3.1 版本开始,它实现了 HandleMessageAdvice ,并以 AbstractHandleMessageAdvice 作为基类,以实现更好的分离。有关更多信息,请参阅 处理消息 Advice。 |