Spring 集成和批处理作业在每个轮询中运行

Question

我不明白 Spring 与 JobLaunchingGateway 行为的集成。我有示例配置：

    public SftpInboundChannelAdapterSpec sftpInboundChannelAdapterSpec() {
        return Sftp.inboundAdapter(ftpFileSessionFactory())
                .preserveTimestamp(true)
                .deleteRemoteFiles(false)
                .remoteDirectory(integrationProperties.getRemoteDirectory())
                .filter(sftpFileListFilter())
                .localDirectory(new File(integrationProperties.getLocalDirectory()));
    }

    public PollerSpec pollerSpec() {
        PollerSpec cron = Pollers.cron(integrationProperties.getPollerCron());
        cron.maxMessagesPerPoll(integrationProperties.getMessagePerPoll());

        return cron;
    }

    @Bean
    public IntegrationFlow sftpInboundFlow() {
        return IntegrationFlows.from(sftpInboundChannelAdapterSpec(), pc -> pc.poller(pollerSpec()))
                .transform(fileMessageToJobRequest())
                .handle(jobLaunchingGateway())
                .handle(message -> {
                    logger.info("Handle message: {}", message.getPayload());
                })
                .get();
    }

    @Bean
    public JobLaunchingGateway jobLaunchingGateway() {
        SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
        simpleJobLauncher.setJobRepository(jobRepository);
        simpleJobLauncher.setTaskExecutor(new SyncTaskExecutor());

        JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(simpleJobLauncher);
        return jobLaunchingGateway;
    }

    private ChainFileListFilter<ChannelSftp.LsEntry> sftpFileListFilter() {
        ChainFileListFilter<ChannelSftp.LsEntry> chainFileListFilter = new ChainFileListFilter<>();
        chainFileListFilter.addFilter(new SftpSimplePatternFileListFilter("*.xlsx"));
        chainFileListFilter.addFilter(new SftpPersistentAcceptOnceFileListFilter(metadataStore(), "INT"));
        return chainFileListFilter;
    }

如果我设置每 1 分钟轮询一次，则每分钟都会创建该作业。我在 MetaDataStore 中没有看到任何新记录。

当我用 .handle(jobLaunchingGateway()) 注释行时，

    @Bean
    public IntegrationFlow sftpInboundFlow() {
        return IntegrationFlows.from(sftpInboundChannelAdapterSpec(), pc -> pc.poller(pollerSpec()))
                .transform(fileMessageToJobRequest())
//                .handle(jobLaunchingGateway())
                .handle(message -> {
                    logger.info("Handle message: {}", message.getPayload());
                })
                .get();
    }

一切都按预期进行。

我期望 SFTP 获取新文件，然后为每个文件创建新作业。

我不明白为什么启用 JobLaunchingGateway 后在 MetaDataStore 中看不到记录。

你能帮我解释一下吗？

我是 Spring 集成和批处理的新手？