|
此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 Spring Integration 6.4.3！ |
轮询器
本节描述了 Spring Integration 中 polling 的工作原理。
轮询消费者
当消息端点（通道适配器）连接到通道并实例化时，它们会生成以下实例之一：
实际实现取决于这些终端节点连接到的通道类型。
一个通道适配器，该适配器连接到实现org.springframework.messaging.SubscribableChannelinterface 会生成一个EventDrivenConsumer.
另一方面，连接到实现org.springframework.messaging.PollableChannel接口（例如QueueChannel） 会生成一个PollingConsumer.
轮询使用者允许 Spring 集成组件主动轮询消息，而不是以事件驱动的方式处理消息。
在许多消息传递方案中，它们代表一个关键的横切关注点。 在 Spring Integration 中，轮询使用者基于同名模式，Gregor Hohpe 和 Bobby Woolf 在 Enterprise Integration Patterns 一书中对此进行了描述。 您可以在该书的网站上找到该模式的描述。
有关轮询使用者配置的更多信息，请参阅消息终端节点。
Pollable 消息源
Spring 集成提供了轮询消费者模式的第二种变体。
使用入站通道适配器时，这些适配器通常由SourcePollingChannelAdapter.
例如，当从远程 FTP 服务器位置检索消息时，FTP 入站通道适配器中描述的适配器配置了 Poller 以定期检索消息。
因此，当组件使用 Poller 进行配置时，生成的实例是以下类型之一：
这意味着轮询器用于入站和出站消息传递方案。 以下是使用 Poller 的一些用例：
-
轮询某些外部系统，例如 FTP 服务器、数据库和 Web 服务
-
轮询内部（可轮询）消息通道
-
轮询内部服务（例如在 Java 类上重复执行方法）
AOP 通知类可以应用于轮询器，在advice-chain，例如用于启动事务的事务通知。
从版本 4.1 开始，PollSkipAdvice。
轮询器使用触发器来确定下一次轮询的时间。
这PollSkipAdvice可用于抑制（跳过）轮询，可能是因为存在一些下游条件会阻止处理消息。
要使用此建议，您必须为其提供PollSkipStrategy.
从版本 4.2.5 开始，SimplePollSkipStrategy。
要使用它，您可以将实例作为 bean 添加到应用程序上下文中，将其注入到PollSkipAdvice，并将其添加到 poller's advice 链中。
要跳过轮询，请调用skipPolls().
要恢复轮询，请调用reset().
版本 4.2 在此领域增加了更多灵活性。
参见 Conditional Pollers。 |
延迟确认轮询消息源
从版本 5.0.1 开始，某些模块提供MessageSource支持将确认推迟到 downstream flow 完成（或将消息移交给另一个线程）的实现。
这目前仅限于AmqpMessageSource和KafkaMessageSource.
使用这些消息源，IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK标头（请参阅MessageHeaderAccessor应用程序接口） 将添加到消息中。
当与可轮询消息源一起使用时，标头的值是AcknowledgmentCallback，如下例所示：
@FunctionalInterface
public interface AcknowledgmentCallback {
void acknowledge(Status status);
boolean isAcknowledged();
void noAutoAck();
default boolean isAutoAck();
enum Status {
/**
* Mark the message as accepted.
*/
ACCEPT,
/**
* Mark the message as rejected.
*/
REJECT,
/**
* Reject the message and requeue so that it will be redelivered.
*/
REQUEUE
}
}并非所有消息源（例如，KafkaMessageSource） 支持REJECT地位。
它的处理方式与ACCEPT.
应用程序可以随时确认消息，如下例所示：
Message<?> received = source.receive();
...
StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
.acknowledge(Status.ACCEPT);如果MessageSource连接到SourcePollingChannelAdapter，当 Poller 线程在下游流完成后返回到适配器时，适配器会检查确认是否已被确认，如果没有，则将其状态设置为ACCEPTit（或REJECT如果流引发异常）。
状态值在AcknowledgmentCallback.Status列举.
Spring 集成提供了MessageSourcePollingTemplate要对MessageSource.
这也负责设置ACCEPT或REJECT在AcknowledgmentCallback当MessageHandlercallback 返回（或引发异常）。
以下示例演示如何使用MessageSourcePollingTemplate:
MessageSourcePollingTemplate template =
new MessageSourcePollingTemplate(this.source);
template.poll(h -> {
...
});在这两种情况下 （SourcePollingChannelAdapter和MessageSourcePollingTemplate），您可以通过调用noAutoAck()在回调上。
如果您将消息交给另一个线程并希望稍后确认，则可以执行此作。
并非所有实现都支持此功能（例如，Apache Kafka 不支持，因为偏移量提交必须在同一线程上执行）。
消息源的条件轮询器
本节介绍如何使用条件 Poller。
背景
Advice对象， 在advice-chain在 Poller 上，通知整个轮询任务（消息检索和处理）。
这些 “around advice” 方法无法访问 poll 的任何上下文 — 只能访问 poll 本身。
如前所述，这对于诸如使任务事务性或由于某些外部条件而跳过轮询等要求来说很好。
如果我们希望根据receive轮询的一部分，或者我们是否想根据条件调整轮询器？对于这些实例， Spring 集成提供了“智能”轮询。
“智能”轮询
版本 5.3 引入了ReceiveMessageAdvice接口。
任何Advice对象advice-chain的 API 版本仅应用于receive()操作-MessageSource.receive()和PollableChannel.receive(timeout).
因此，它们只能应用于SourcePollingChannelAdapter或PollingConsumer.
此类实现以下方法：
-
beforeReceive(Object source)此方法在Object.receive()方法。 它允许您检查和重新配置源。 返回false取消此轮询（类似于PollSkipAdvice前面提到过）。 -
Message<?> afterReceive(Message<?> result, Object source)该方法在receive()方法。 同样，您可以重新配置源或采取任何作（可能取决于结果，可以是null如果源没有创建消息）。 您甚至可以返回不同的消息
|
线程安全
如果 |
|
Advice Chain 订购
您应该了解在初始化期间如何处理通知链。 |
SimpleActiveIdleReceiveMessageAdvice
此建议是ReceiveMessageAdvice.
当与DynamicPeriodicTrigger，它会调整轮询频率，具体取决于上一次轮询是否导致消息。
轮询器还必须具有对相同DynamicPeriodicTrigger.
|
重要： 异步切换
SimpleActiveIdleReceiveMessageAdvice根据receive()结果。
这仅在 poller 线程上调用通知时有效。
如果 Poller 具有task-executor.
要在轮询结果后使用异步作时使用此建议，请稍后执行异步切换，可能通过使用ExecutorChannel. |
CompoundTriggerAdvice
此建议允许根据 poll 是否返回消息来选择两个触发器之一。
考虑一个使用CronTrigger.CronTrigger实例是不可变的，因此一旦构建就无法更改。
考虑这样一个使用案例：我们希望使用 cron 表达式每小时触发一次轮询，但如果未收到消息，则每分钟轮询一次，并在检索到消息时恢复为使用 cron 表达式。
建议（和 poller）使用CompoundTrigger为此目的。
触发器的primarytrigger 可以是CronTrigger.
当通知检测到未收到任何消息时，它会将辅助触发器添加到CompoundTrigger.
当CompoundTrigger实例的nextExecutionTime方法，它会委托给辅助触发器（如果存在）。
否则，它将委托给主触发器。
轮询器还必须具有对相同CompoundTrigger.
以下示例显示了回退到每分钟的每小时 cron 表达式的配置：
<int:inbound-channel-adapter channel="nullChannel" auto-startup="false">
<bean class="org.springframework.integration.endpoint.PollerAdviceTests.Source" />
<int:poller trigger="compoundTrigger">
<int:advice-chain>
<bean class="org.springframework.integration.aop.CompoundTriggerAdvice">
<constructor-arg ref="compoundTrigger"/>
<constructor-arg ref="secondary"/>
</bean>
</int:advice-chain>
</int:poller>
</int:inbound-channel-adapter>
<bean id="compoundTrigger" class="org.springframework.integration.util.CompoundTrigger">
<constructor-arg ref="primary" />
</bean>
<bean id="primary" class="org.springframework.scheduling.support.CronTrigger">
<constructor-arg value="0 0 * * * *" /> <!-- top of every hour -->
</bean>
<bean id="secondary" class="org.springframework.scheduling.support.PeriodicTrigger">
<constructor-arg value="60000" />
</bean>|
重要： 异步切换
CompoundTriggerAdvice根据receive()结果。
这仅在 poller 线程上调用通知时有效。
如果 Poller 具有task-executor.
要在轮询结果后使用异步作时使用此建议，请稍后执行异步切换，可能通过使用ExecutorChannel. |
仅限 MessageSource 的建议
某些建议可能仅适用于MessageSource.receive()而且它们对PollableChannel.
为此，一个MessageSourceMutator接口（ReceiveMessageAdvice） 仍然存在。
有关更多信息，请参见入站通道适配器：轮询多个服务器和目录。