收到消息后如何停止轮询？弹簧集成

Question

我想轮询目录中的文件并在找到文件后停止轮询。我对 Spring 框架非常陌生，其中很多内容仍然非常令人困惑。经过一些研究后，我发现了几种方法可以做到这一点，但没有任何运气。

其中一种方法是使用控制总线，如下所示。然而，轮询似乎在 2 秒后就停止了。我不确定如何包含仅在收到文件时停止的条件。

另一种方法是使用此处回答的“智能轮询” 。答案中的链接很旧，但它指向此处的官方 Spring 文档：Smart Polling。通过这篇文章，我了解了AbstractMessageSourceAdvice和SimpleActiveIdleMessageSourceAdvice。后者似乎适合我的目标，并且是最容易实现的，所以我决定尝试一下。我的代码如下：

集成配置.java

package com.example.springexample;

import java.io.File;

import org.aopalliance.aop.Advice;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.aop.SimpleActiveIdleMessageSourceAdvice;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.filters.SimplePatternFileListFilter;
import org.springframework.integration.util.DynamicPeriodicTrigger;
import org.springframework.messaging.MessageChannel;

@Configuration
@EnableIntegration
public class IntegrationConfig {

    @Bean
    public IntegrationFlow advised() {
        return IntegrationFlows.from("fileInputChannel")
                .handle("runBatchScript", "run", c -> c.advice(stopPollingAdvice()))
                .get();
    }

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

    @Bean
    @InboundChannelAdapter(value = "fileInputChannel", poller = @Poller(fixedDelay = "1000"))
    public MessageSource<File> fileReadingMessageSource() {
        FileReadingMessageSource source = new FileReadingMessageSource();
        source.setDirectory(new File("."));
        source.setFilter(new SimplePatternFileListFilter("*.bat"));
        return source;
    }

    @Bean
    public RunBatchScript runBatchScript() {
        return new RunBatchScript();
    }

    @Bean
    public Advice stopPollingAdvice() {
        DynamicPeriodicTrigger trigger = new DynamicPeriodicTrigger(10000);
        SimpleActiveIdleMessageSourceAdvice advice = new SimpleActiveIdleMessageSourceAdvice(trigger);
        advice.setActivePollPeriod(60000);
        return advice;
    }
}

RunBatchScript.java

package com.example.springexample;

import java.io.IOException;
import java.util.Date;
import java.util.logging.Logger;

public class RunBatchScript {

    Logger logger = Logger.getLogger(RunBatchScript.class.getName());

    public void run() throws IOException {
        logger.info("Running the batch script at " + new Date());
        Runtime.getRuntime().exec("cmd.exe /c simplebatchscript.bat");
        logger.info("Finished running the batch script at " + new Date());
    }
}

SpringExampleApplication.java

package com.example.springexample;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SpringExampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringExampleApplication.class, args);
    }

}

我用这个和这个作为我的代码的基础。然而，它似乎不起作用，因为轮询器仍然每 1 秒轮询一次，而不是新的 10 秒或 60 秒。此外，我不确定如何真正停止轮询器。我尝试将其放入null构造函数中SimpleActiveIdleMessageSource，但它只返回NullPointerException。

我运行应用程序时的输出：

2020-03-15 13:57:46.081  INFO 37504 --- [ask-scheduler-1] c.example.springexample.RunBatchScript   : Running the batch script at Sun Mar 15 13:57:46 SRET 2020
2020-03-15 13:57:46.084  INFO 37504 --- [ask-scheduler-1] c.example.springexample.RunBatchScript   : Finished running the batch script at Sun Mar 15 13:57:46 SRET 2020
2020-03-15 13:57:47.085  INFO 37504 --- [ask-scheduler-2] c.example.springexample.RunBatchScript   : Running the batch script at Sun Mar 15 13:57:47 SRET 2020
2020-03-15 13:57:47.087  INFO 37504 --- [ask-scheduler-2] c.example.springexample.RunBatchScript   : Finished running the batch script at Sun Mar 15 13:57:47 SRET 2020
2020-03-15 13:57:48.089  INFO 37504 --- [ask-scheduler-1] c.example.springexample.RunBatchScript   : Running the batch script at Sun Mar 15 13:57:48 SRET 2020
2020-03-15 13:57:48.092  INFO 37504 --- [ask-scheduler-1] c.example.springexample.RunBatchScript   : Finished running the batch script at Sun Mar 15 13:57:48 SRET 2020
2020-03-15 13:57:49.093  INFO 37504 --- [ask-scheduler-3] c.example.springexample.RunBatchScript   : Running the batch script at Sun Mar 15 13:57:49 SRET 2020
2020-03-15 13:57:49.096  INFO 37504 --- [ask-scheduler-3] c.example.springexample.RunBatchScript   : Finished running the batch script at Sun Mar 15 13:57:49 SRET 2020

非常感谢任何有关某些代码的帮助。

Answer 1

您应该SimpleActiveIdleMessageSourceAdvice申请@InboundChannelAdapter. 另外， 的触发器SimpleActiveIdleMessageSourceAdvice应该与用于轮询文件的触发器相同：

    @Bean
    @EndpointId("fileInboundChannelAdapter")
    @InboundChannelAdapter(value = "fileInputChannel", poller = @Poller("fileReadingMessageSourcePollerMetadata"))
    public MessageSource<File> fileReadingMessageSource() {
        FileReadingMessageSource source = new FileReadingMessageSource();
        source.setDirectory(new File("."));
        source.setFilter(new SimplePatternFileListFilter("*.bat"));
        return source;
    }

    @Bean
    public PollerMetadata fileReadingMessageSourcePollerMetadata() {
        PollerMetadata meta = new PollerMetadata();

        DynamicPeriodicTrigger trigger = new DynamicPeriodicTrigger(1000);

        SimpleActiveIdleMessageSourceAdvice advice = new SimpleActiveIdleMessageSourceAdvice(trigger);
        advice.setActivePollPeriod(60000);

        meta.setTrigger(trigger);
        meta.setAdviceChain(List.of(advice));
        meta.setMaxMessagesPerPoll(1);
        return meta;
    }

请注意，SimpleActiveIdleMessageSourceAdvice只需更改下次轮询文件的时间即可。您可以将其设置为一个非常大的数字，例如几千年后，这可以以某种方式实现您的意图，在您的一生中不再轮询文件。但轮询文件的调度程序线程仍然处于活动状态。

如果您确实也想关闭该调度程序线程，则可以向控制总线发送关闭信号。

首先定义一个控制总线：

    @Bean
    public IntegrationFlow controlBusFlow() {
        return IntegrationFlows.from("controlBus")
                  .controlBus()
                  .get();
    }

然后实现AbstractMessageSourceAdvice在轮询文件后向控制总线发送关闭信号：

@Service
public class StopPollingAdvice extends AbstractMessageSourceAdvice{

    @Lazy
    @Qualifier("controlBus")
    @Autowired
    private MessageChannel controlBusChannel;


    @Override
    public boolean beforeReceive(MessageSource<?> source) {
        return super.beforeReceive(source);
    }

    @Override
    public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
        Message operation = MessageBuilder.withPayload("@fileInboundChannelAdapter.stop()").build();
        controlBusChannel.send(operation);
        return result;
    }
}

并将该PollerMetadata轮询文件更改为：

@Bean
public PollerMetadata fileReadingMessageSourcePollerMetadata(StopPollingAdvice stopPollingAdvice) {
    PollerMetadata meta = new PollerMetadata(); 
    meta.setTrigger(new PeriodicTrigger(1000));
    meta.setAdviceChain(List.of(stopPollingAdvice));
    meta.setMaxMessagesPerPoll(1);
    return meta;
}