延迟器
延迟器是一个简单的端点,它允许消息流延迟一定的时间间隔。
当消息延迟时,原始发件人不会阻止。
相反,延迟的消息是使用org.springframework.scheduling.TaskScheduler
在 delay 过后发送到 output 通道。
这种方法即使对于相当长的延迟也是可扩展的,因为它不会导致大量阻塞的发送方线程。
相反,在典型情况下,线程池用于实际执行释放消息。
本节包含配置延迟器的几个示例。
配置 Delayer
这<delayer>
元素用于延迟两个消息通道之间的消息流。
与其他终端节点一样,您可以提供 'input-channel' 和 'output-channel' 属性,但延迟器还具有 'default-delay' 和 'expression' 属性(以及 'expression' 元素),用于确定每条消息应延迟的毫秒数。
以下示例将所有消息延迟 3 秒:
<int:delayer id="delayer" input-channel="input"
default-delay="3000" output-channel="output"/>
如果需要确定每条消息的延迟,还可以使用'expression'属性提供 SPEL 表达式,如以下表达式所示:
-
Java DSL
-
Kotlin DSL
-
Java
-
XML
@Bean
public IntegrationFlow flow() {
return IntegrationFlow.from("input")
.delay(d -> d
.messageGroupId("delayer.messageGroupId")
.defaultDelay(3_000L)
.delayExpression("headers['delay']"))
.channel("output")
.get();
}
@Bean
fun flow() =
integrationFlow("input") {
delay {
messageGroupId("delayer.messageGroupId")
defaultDelay(3000L)
delayExpression("headers['delay']")
}
channel("output")
}
@ServiceActivator(inputChannel = "input")
@Bean
public DelayHandler delayer() {
DelayHandler handler = new DelayHandler("delayer.messageGroupId");
handler.setDefaultDelay(3_000L);
handler.setDelayExpressionString("headers['delay']");
handler.setOutputChannelName("output");
return handler;
}
<int:delayer id="delayer" input-channel="input" output-channel="output"
default-delay="3000" expression="headers['delay']"/>
在前面的示例中,仅当表达式的计算结果为给定入站消息的 null 时,三秒延迟才适用。
如果只想将延迟应用于具有表达式评估有效结果的消息,则可以使用 'default-delay'0
(默认值)。
对于延迟为0
(或更少),则消息会立即在调用线程上发送。
XML 解析器使用消息组 ID<beanName>.messageGroupId . |
延迟处理程序支持表示以毫秒为单位的间隔(任何Object 谁的toString() method 生成一个可以解析为Long ) 以及java.util.Date 表示绝对时间的实例。
在第一种情况下,毫秒从当前时间开始计算(例如,值5000 将消息从延迟器收到消息的时间起至少延迟 5 秒)。
使用Date 实例,则消息在Date 对象。
等于非正延迟或过去的 Date 的值不会导致延迟。
相反,它被直接发送到原始发送方线程上的 output 通道。
如果表达式计算结果不是Date ,并且不能解析为Long ,则默认延迟(如果有 — 默认值为0 ) 应用。 |
表达式计算可能会因各种原因(包括无效的表达式或其他条件)而引发计算异常。
默认情况下,此类异常将被忽略(尽管记录在 DEBUG 级别),并且延迟器会回退到默认延迟(如果有)。
您可以通过设置ignore-expression-failures 属性。
默认情况下,此属性设置为true 延迟器行为如前所述。
但是,如果你不希望忽略表达式计算异常并将其抛给延迟器的调用者,请将ignore-expression-failures 属性设置为false . |
在前面的示例中,延迟表达式指定为
因此,如果可能会省略标头,并且你想要回退到默认延迟,则使用索引器语法而不是点属性访问器语法通常更有效(并且建议使用),因为检测 null 比捕获异常更快。 |
延迟器委托给 Spring 的TaskScheduler
抽象化。
延迟器使用的默认调度程序是ThreadPoolTaskScheduler
实例。
请参阅配置 Task Scheduler。
如果要委托给不同的 scheduler,可以通过 delayer 元素的 'scheduler' 属性提供引用,如下例所示:
<int:delayer id="delayer" input-channel="input" output-channel="output"
expression="headers.delay"
scheduler="exampleTaskScheduler"/>
<task:scheduler id="exampleTaskScheduler" pool-size="3"/>
如果您配置外部ThreadPoolTaskScheduler 中,您可以设置waitForTasksToCompleteOnShutdown = true 在此属性上。
它允许在应用程序关闭时成功完成已经处于执行状态(释放消息)的 'delay' 任务。
在 Spring Integration 2.2 之前,这个属性在<delayer> 元素,因为DelayHandler 可以在后台创建自己的调度器。
从 2.2 开始,延迟器需要一个外部调度器实例,并且waitForTasksToCompleteOnShutdown 已删除。
您应该使用调度程序自己的配置。 |
ThreadPoolTaskScheduler 具有属性errorHandler ,它可以注入一些org.springframework.util.ErrorHandler .
此处理程序允许处理Exception 从发送延迟消息的计划任务的线程中。
默认情况下,它使用org.springframework.scheduling.support.TaskUtils$LoggingErrorHandler ,您可以在日志中看到堆栈跟踪。
您可能需要考虑使用org.springframework.integration.channel.MessagePublishingErrorHandler ,它会发送一个ErrorMessage 转换为error-channel ,从失败邮件的标头或默认的error-channel .
此错误处理在事务回滚(如果存在)后执行。
请参阅 发布失败。 |
Delayer 和 Message Store
这DelayHandler
将延迟的消息持久化到提供的MessageStore
.
(“groupId”基于<delayer>
元素。
另请参阅DelayHandler.setMessageGroupId(String)
.)
延迟消息将从MessageStore
通过紧接DelayHandler
将消息发送到output-channel
.
如果提供的MessageStore
是持久性的(例如JdbcMessageStore
),它提供了在应用程序关闭时不丢失消息的能力。
应用程序启动后,DelayHandler
从MessageStore
并根据消息的原始到达时间(如果延迟是数字)重新安排它们并带有延迟。
对于延迟标头为Date
那Date
在重新安排时使用。
如果延迟的消息保留在MessageStore
超过其 'delay' ,它会在启动后立即发送。
这messageGroupId
是必需的,不能依赖于DelayHandler
bean 名称。
这样,在应用程序重新启动后,DelayHandler
可能会获得新生成的 bean 名称。
因此,延迟的消息可能会因重新调度而丢失,因为它们的组不再由应用程序管理。
这<delayer>
可以使用两个互斥元素之一进行扩充:<transactional>
和<advice-chain>
.
这List
的 AOP 建议应用于代理的内部DelayHandler.ReleaseMessageHandler
,它负责在延迟后在Thread
的计划任务。
例如,当下游消息流抛出异常并且ReleaseMessageHandler
将回滚。
在这种情况下,延迟消息将保留在持久MessageStore
.
您可以使用任何自定义org.aopalliance.aop.Advice
在<advice-chain>
.
这<transactional>
元素定义一个只有事务性建议的简单通知链。
以下示例显示了advice-chain
在<delayer>
:
<int:delayer id="delayer" input-channel="input" output-channel="output"
expression="headers.delay"
message-store="jdbcMessageStore">
<int:advice-chain>
<beans:ref bean="customAdviceBean"/>
<tx:advice>
<tx:attributes>
<tx:method name="*" read-only="true"/>
</tx:attributes>
</tx:advice>
</int:advice-chain>
</int:delayer>
这DelayHandler
可以导出为 JMXMBean
使用托管式作 (getDelayedMessageCount
和reschedulePersistedMessages
),这允许在运行时重新调度延迟的持久化消息 — 例如,如果TaskScheduler
之前已停止。
这些作可以通过Control Bus
命令,如下例所示:
Message<String> delayerReschedulingMessage =
MessageBuilder.withPayload("'delayer.handler'.reschedulePersistedMessages").build();
controlBusChannel.send(delayerReschedulingMessage);
有关消息存储、JMX 和控制总线的更多信息,请参阅系统管理。 |
从版本 5.3.7 开始,如果在将消息存储到MessageStore
,则发布任务计划在TransactionSynchronization.afterCommit()
回调。
这对于防止争用情况是必要的,在这种情况下,计划的发布可能会在事务提交之前运行,并且找不到消息。
在这种情况下,消息将在延迟后或事务提交后(以较晚者为准)发布。
发布失败
从版本 5.0.8 开始,延迟器上有两个新属性:
-
maxAttempts
(默认 5) -
retryDelay
(默认 1 秒)
当消息被释放时,如果下游流失败,将在retryDelay
.
如果maxAttempts
,则消息将被丢弃(除非发布是事务性的,在这种情况下,消息将保留在存储中,但不会再计划发布,直到应用程序重新启动,或者reschedulePersistedMessages()
方法,如上所述)。
此外,您还可以配置delayedMessageErrorChannel
;当发布失败时,ErrorMessage
发送到该通道,但 exception 作为有效负载,并且具有originalMessage
财产。
这ErrorMessage
包含标头IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT
包含当前计数。
如果错误流使用错误消息并正常退出,则不会执行进一步的作;如果发布是事务性的,则将提交事务并从存储中删除消息。
如果错误流引发异常,则将重试该版本,直到maxAttempts
如上所述。