如何停止或更改 Spring 集成轮询器的延迟

Question

我正在使用 Spring 集成来使用以下配置从目录中读取文件。但是，我希望在找到任何文件后停止轮询，直到服务不再重新启动为止。有什么办法可以在运行时更改轮询器延迟或在运行时更改 start/stop 轮询器？

@Bean
public MessageChannel fileInputChannel() {
    return new DirectChannel();
}

@Bean
@InboundChannelAdapter(channel = "fileInputChannel", poller = @Poller(cron = "0 0/10 19-23,0-6 ? * *", maxMessagesPerPoll = "1"))
public MessageSource<File> fileReadingMessageSource() {
    FileReadingMessageSource source = new FileReadingMessageSource();
    File directory = new File(localFtpDirectory);
    if (clearLocalDir && directory.isDirectory() && directory.exists()) {
        LOG.info("Clear directory {} on startup of service", directory);
        Arrays.stream(directory.listFiles()).forEach(File::delete);
    }
    source.setDirectory(directory);
    source.setFilter(new LastModifiedFileFilter(remoteFileFilter));
    return source;
}

@Bean
@ServiceActivator(inputChannel = "fileInputChannel")
public MessageHandler fileHandler() {
    return new MessageHandlerService();
}

Answer 1

有这个注释可以与 @InboundChannelAdapter:

一起使用

/**
 * When used alongside an EIP annotation (and no {@code @Bean}), specifies the bean name of
 * the consumer bean with the handler bean being {@code id.handler} (for a consuming
 * endpoint) or {@code id.source} for a message source (e.g. inbound channel adapter).
 * <p>
 * When there is also a {@code @Bean} annotation, this is the name of the consumer or
 * source polling bean (the handler or source gets the normal {@code @Bean} name). When
 * using on a {@code MessageHandler @Bean}, it is recommended to name the bean
 * {@code foo.handler} when using {@code @EndpointId("foo"}. This will align with
 * conventions in the framework. Similarly, for a message source, use
 * {@code @Bean("bar.source"} and {@code @EndpointId("bar")}.
 * <p>
 * <b>This is not allowed if there are multiple EIP annotations on the same method.</b>
 *
 * @author Gary Russell
 *
 * @since 5.0.4
 */
@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface EndpointId {

并且有一个 Cotrol 总线 EIP 实现：https://docs.spring.io/spring-integration/docs/5.0.7.RELEASE/reference/html/system-management-chapter.html#control-bus。

有了它，您可以随时发送 start()/stop() 命令消息。

Answer 2

我有类似的要求，但对接受的答案不满意。这里提到的 start/stop 功能是由 Lifecycle/SmartLifecycle 契约定义的，我能看到的典型场景是应用程序关闭或应用程序上下文刷新。简而言之，当我天真地尝试使用它时，我确实很快就遇到了问题。

我的要求是暂停轮询器，然后恢复。这可以通过 advice 轻松实现，并且非常容易实现。然而，获得这些知识并不容易，我最终进行了框架源代码调查（实际上我可能忽略了文档中的某些内容）。

    @Bean
    public MessagePollingControlAdvice messagePollingControlAdvice() {
        return new MessagePollingControlAdvice();
    }

    //call this method for your DSL bridge configuration etc..
    Consumer<GenericEndpointSpec<BridgeHandler>> pollerConfiguration() {
        return b -> b.poller(pollerFactory -> pollerFactory.advice(messagePollingControlAdvice()));
    }
   

   public static class MessagePollingControlAdvice implements ReceiveMessageAdvice {
    private volatile boolean pollingActive = false;

    @Override
    public boolean beforeReceive(Object source) {
        return pollingActive;
    }

    @Override
    public Message<?> afterReceive(Message<?> result, Object source) {
        return result;
    }

    public boolean isPollingActive() {
        return pollingActive;
    }

    //call this method from whatever place in your code to activate/deactivate poller
    public void setPollingActive(boolean pollingActive) {
        this.pollingActive = pollingActive;
    }
   }

如何停止或更改 Spring 集成轮询器的延迟

How to stop OR change delay of Spring Integration Poller

spring-integration

spring-integration-sftp