轮询器 (Poller) _ Spring Integration6.2.9中文文档(官方文档中文翻译)|Spring 教程 —

对于最新的稳定版本，请使用 Spring Integration 6.3.4！spring-doc.cn

本节描述了 Spring Integration 中 polling 的工作原理。spring-doc.cn

轮询消费者

当消息端点（通道适配器）连接到通道并实例化时，它们会生成以下实例之一：spring-doc.cn

实际实现取决于这些终端节点连接到的通道类型。连接到实现 org.springframework.messaging.SubscribableChannel 接口的通道适配器会生成一个 . 另一方面，连接到实现 org.springframework.messaging.PollableChannel 接口（例如 a ）的通道适配器会生成一个 .EventDrivenConsumerQueueChannelPollingConsumerspring-doc.cn

轮询使用者允许 Spring 集成组件主动轮询消息，而不是以事件驱动的方式处理消息。spring-doc.cn

在许多消息传递方案中，它们代表一个关键的横切关注点。在 Spring Integration 中，轮询使用者基于同名模式，Gregor Hohpe 和 Bobby Woolf 在 Enterprise Integration Patterns 一书中对此进行了描述。您可以在该书的网站上找到该模式的描述。spring-doc.cn

有关轮询使用者配置的更多信息，请参阅消息终端节点。spring-doc.cn

Pollable 消息源

Spring 集成提供了轮询消费者模式的第二种变体。使用入站通道适配器时，这些适配器通常由 . 例如，当从远程 FTP 服务器位置检索消息时，FTP 入站通道适配器中描述的适配器配置了 Poller 以定期检索消息。因此，当组件使用 Poller 进行配置时，生成的实例是以下类型之一：SourcePollingChannelAdapterspring-doc.cn

这意味着轮询器用于入站和出站消息传递方案。以下是使用 Poller 的一些用例：spring-doc.cn

轮询某些外部系统，例如 FTP 服务器、数据库和 Web 服务spring-doc.cn
轮询内部（可轮询）消息通道spring-doc.cn
轮询内部服务（例如在 Java 类上重复执行方法）spring-doc.cn

AOP 建议类可以应用于 Poller ，例如用于启动事务的事务通知。从版本 4.1 开始，提供了 a。轮询器使用触发器来确定下一次轮询的时间。这可用于抑制（跳过）轮询，可能是因为存在一些下游条件会阻止消息被处理。要使用此建议，您必须为其提供 . 从版本 4.2.5 开始，提供了 a。要使用它，你可以将实例作为 bean 添加到应用程序上下文中，将其注入到，并将其添加到 Poller 的建议链中。要跳过轮询，请调用。要恢复轮询，请调用。版本 4.2 在此领域增加了更多灵活性。参见 Conditional Pollers。advice-chainPollSkipAdvicePollSkipAdvicePollSkipStrategySimplePollSkipStrategyPollSkipAdviceskipPolls()reset()

本章仅对轮询使用者以及它们如何适应消息通道（请参见消息通道）和通道适配器（请参见通道适配器）的概念进行简要概述。有关消息收发终端节点的一般信息，特别是轮询使用者的更多信息，请参阅消息终端节点。spring-doc.cn

延迟确认轮询消息源

从版本 5.0.1 开始，某些模块提供的实现支持将确认推迟到下游流完成（或将消息移交给另一个线程）。目前仅限于和 .MessageSourceAmqpMessageSourceKafkaMessageSourcespring-doc.cn

使用这些消息源，标头（请参阅 MessageHeaderAccessor API）将添加到消息中。当与可轮询消息源一起使用时，标头的值是的实例，如下例所示：IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACKAcknowledgmentCallbackspring-doc.cn

@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

    }

}

并非所有消息源（例如 a ）都支持该状态。它的处理方式与相同。KafkaMessageSourceREJECTACCEPTspring-doc.cn

应用程序可以随时确认消息，如下例所示：spring-doc.cn

Message<?> received = source.receive();

...

StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
        .acknowledge(Status.ACCEPT);

如果 the 被连接到 a ，当 poller 线程在下游流完成后返回到适配器时，适配器会检查确认是否已被确认，如果没有，则将其状态设置为它（或者 flow 是否引发异常）。状态值在 AcknowledgmentCallback.Status 枚举中定义。MessageSourceSourcePollingChannelAdapterACCEPTREJECTspring-doc.cn

Spring Integration 提供了对 . 这也负责设置或回调何时返回（或引发异常）。以下示例演示如何使用：MessageSourcePollingTemplateMessageSourceACCEPTREJECTAcknowledgmentCallbackMessageHandlerMessageSourcePollingTemplatespring-doc.cn

MessageSourcePollingTemplate template =
    new MessageSourcePollingTemplate(this.source);
template.poll(h -> {
    ...
});

在这两种情况下（和），您都可以通过调用回调来禁用 auto ack/nack。如果您将消息交给另一个线程并希望稍后确认，则可以执行此操作。并非所有实现都支持此功能（例如，Apache Kafka 不支持，因为偏移量提交必须在同一线程上执行）。SourcePollingChannelAdapterMessageSourcePollingTemplatenoAutoAck()spring-doc.cn

消息源的条件轮询器

本节介绍如何使用条件 Poller。spring-doc.cn

背景

Advice对象，在 ON POLLER 中，通知整个轮询任务（消息检索和处理）。这些 “around advice” 方法无法访问 poll 的任何上下文 — 只能访问 poll 本身。如前所述，这对于诸如使任务事务性或由于某些外部条件而跳过轮询等要求来说很好。如果我们希望根据 poll 部分的结果采取一些行动，或者我们想根据条件调整 Poller 怎么办？对于这些实例， Spring 集成提供了“智能”轮询。advice-chainreceivespring-doc.cn

“智能”轮询

版本 5.3 引入了该界面。中实现此接口的任何对象仅应用于操作 - 和。因此，它们只能应用于或。此类实现以下方法：ReceiveMessageAdviceAdviceadvice-chainreceive()MessageSource.receive()PollableChannel.receive(timeout)SourcePollingChannelAdapterPollingConsumerspring-doc.cn

beforeReceive(Object source)该方法在该方法之前调用。它允许您检查和重新配置源。返回将取消此轮询（类似于前面提到的）。Object.receive()falsePollSkipAdvicespring-doc.cn
Message<?> afterReceive(Message<?> result, Object source)该方法在该方法之后调用。同样，您可以重新配置源或执行任何操作（可能取决于结果，如果源没有创建消息，则可能会有所不同）。您甚至可以返回不同的消息receive()nullspring-doc.cn

线程安全

如果 an 更改了源，则不应使用 . 如果 an 更改了源，则此类更改不是线程安全的，并且可能会导致意外结果，尤其是对于高频轮询器。如果需要并发处理 poll 结果，请考虑使用下游，而不是向 Poller 添加执行程序。AdviceTaskExecutorAdviceExecutorChannelspring-doc.cn

Advice Chain 订购

您应该了解在初始化期间如何处理通知链。未实现的对象将应用于整个轮询过程，并且所有对象都首先按顺序调用，然后再调用 any 。然后，围绕 source 方法按顺序调用对象。例如，如果您有 objects 、 where 和 are ，则对象将按以下顺序应用：。此外，如果源已经是，则会在任何现有对象之后调用。如果您想更改订单，您必须自己连接代理。AdviceReceiveMessageAdviceReceiveMessageAdviceReceiveMessageAdvicereceive()Advicea, b, c, dbdReceiveMessageAdvicea, c, b, dProxyReceiveMessageAdviceAdvicespring-doc.cn

`SimpleActiveIdleReceiveMessageAdvice`

此建议是 . 当与一起使用时，它会根据上一次轮询是否生成消息来调整轮询频率。 poller 还必须具有对 same 的引用。ReceiveMessageAdviceDynamicPeriodicTriggerDynamicPeriodicTriggerspring-doc.cn

重要：异步切换

SimpleActiveIdleReceiveMessageAdvice根据结果修改触发器。这仅在 poller 线程上调用通知时有效。如果 Poller 具有 . 要在轮询结果后使用异步操作时使用此建议，请稍后执行异步切换，可能使用 .receive()task-executorExecutorChannel

`CompoundTriggerAdvice`

此建议允许根据 poll 是否返回消息来选择两个触发器之一。考虑一个使用 . 实例是不可变的，因此一旦构建就无法更改。考虑这样一个使用案例：我们希望使用 cron 表达式每小时触发一次轮询，但如果未收到任何消息，则每分钟轮询一次，并在检索到消息时恢复为使用 cron 表达式。CronTriggerCronTriggerspring-doc.cn

为此，建议（和 poller）使用 a。触发器的触发器可以是 . 当通知检测到未收到任何消息时，它会将辅助触发器添加到 . 调用实例的方法时，它会委托给辅助触发器（如果存在）。否则，它将委托给主触发器。CompoundTriggerprimaryCronTriggerCompoundTriggerCompoundTriggernextExecutionTimespring-doc.cn

poller 还必须具有对 same 的引用。CompoundTriggerspring-doc.cn

以下示例显示了回退到每分钟的每小时 cron 表达式的配置：spring-doc.cn

<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根据结果修改触发器。这仅在 poller 线程上调用通知时有效。如果 Poller 具有 . 要在轮询结果后使用异步操作时使用此建议，请稍后执行异步切换，可能使用 .receive()task-executorExecutorChannel

仅限 MessageSource 的建议

有些建议可能仅适用于，而对则没有意义。为此，仍然存在一个接口（的扩展）。有关更多信息，请参见入站通道适配器：轮询多个服务器和目录。MessageSource.receive()PollableChannelMessageSourceMutatorReceiveMessageAdvicespring-doc.cn

线程安全

Advice Chain 订购

重要：异步切换