spring - 收到消息后如何停止轮询？ Spring 集成

Question

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

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

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

IntegrationConfig.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);
    }

}

我用了this和 this作为我的代码的基础。然而，它似乎不起作用，因为轮询器仍然每 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;
}