使用 Spring Batch 集成为 AWS S3 中的每个新文件启动 JobLaunchRequest

问题描述

我正在关注文档：Spring Batch Integration 结合 Integration AWS 以汇集 AWS S3。

但是在某些情况下，每个文件的批处理执行不起作用。

AWS S3 池工作正常，所以当我放置一个新文件或当我启动应用程序并且存储桶中有文件时，应用程序与本地目录同步：

    @Bean
    public S3SessionFactory s3SessionFactory(AmazonS3 pAmazonS3) {
        return new S3SessionFactory(pAmazonS3);
    }

    @Bean
    public S3InboundFileSynchronizer s3InboundFileSynchronizer(S3SessionFactory ps3SessionFactory) {
        S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(ps3SessionFactory);
        synchronizer.setPreserveTimestamp(true);
        synchronizer.setDeleteRemoteFiles(false);
        synchronizer.setRemoteDirectory("remote-bucket");
        //synchronizer.setFilter(new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),"simpleMetadataStore"));
        return synchronizer;
    }

    @Bean
    @InboundChannelAdapter(value = IN_CHANNEL_NAME,poller = @Poller(fixedDelay = "30"))
    public S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource(
            S3InboundFileSynchronizer ps3InboundFileSynchronizer) {
        S3InboundFileSynchronizingMessageSource messageSource = new S3InboundFileSynchronizingMessageSource(ps3InboundFileSynchronizer);
        messageSource.setautocreateLocalDirectory(true);
        messageSource.setLocalDirectory(new FileSystemResource("files").getFile());
        //messageSource.setLocalFilter(new FileSystemPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),"fsSimpleMetadataStore"));
        return messageSource;
    }

    @Bean("s3filesChannel")
    public PollableChannel s3FilesChannel() {
        return new QueueChannel();
    }

我按照教程制作了FileMessagetoJobRequest 我不会把代码放在这里，因为它和文档一样

所以我创建了 bean IntegrationFlow 和 FileMessagetoJobRequest：

    @Bean
    public IntegrationFlow integrationFlow(
            S3InboundFileSynchronizingMessageSource ps3InboundFileSynchronizingMessageSource) {
        return IntegrationFlows.from(ps3InboundFileSynchronizingMessageSource,c -> c.poller(Pollers.fixedrate(1000).maxMessagesPerPoll(1)))
                .transform(fileMessagetoJobRequest())
                .handle(jobLaunchingGateway())
                .log(LoggingHandler.Level.WARN,"headers.id + ': ' + payload")
                .get();
    }

    @Bean
    public FileMessagetoJobRequest fileMessagetoJobRequest() {
        FileMessagetoJobRequest fileMessagetoJobRequest = new FileMessagetoJobRequest();
        fileMessagetoJobRequest.setFileParameterName("input.file.name");
        fileMessagetoJobRequest.setJob(delimitedFileJob);
        return fileMessagetoJobRequest;
    }

所以在 JobLaunchingGateway 中，我认为是问题所在：

如果我是这样创建的：

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

        return jobLaunchingGateway;
    }

案例 1（应用程序启动时桶为空）：

我在 AWS S3 中上传了一个新文件；
池化工作并且文件出现在本地目录中；
但是转换/作业没有被触发；

案例 2（当应用程序启动时，Bucket 已经有一个文件）：

作业已启动：

2021-01-12 13:32:34.451  INFO 1955 --- [ask-scheduler-1] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=arquivoDelimitadoJob]] launched with the following parameters: [{input.file.name=files/FILE1.csv}]
2021-01-12 13:32:34.524  INFO 1955 --- [ask-scheduler-1] o.s.batch.core.job.SimpleStepHandler     : Executing step: [delimitedFileJob]

如果我在 S3 中添加第二个文件，则不会像情况 1 那样启动作业。

案例 3（Bucket 有多个文件）：

文件在本地目录中正确同步
但该作业仅对最后一个文件执行一次。

所以在 docs 之后，我将网关更改为：

    @Bean
    @ServiceActivator(inputChannel = IN_CHANNEL_NAME,poller = @Poller(fixedrate="1000"))
    public JobLaunchingGateway jobLaunchingGateway() {
        SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
        simpleJobLauncher.setJobRepository(jobRepository);
        simpleJobLauncher.setTaskExecutor(new SyncTaskExecutor());

        //JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(jobLauncher());
        JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(simpleJobLauncher);
        //jobLaunchingGateway.setoutputChannel(replyChannel());
        jobLaunchingGateway.setoutputChannel(s3FilesChannel());
        return jobLaunchingGateway;
    }

使用这个新的网关实现，如果我在 S3 中放置一个新文件，应用程序会做出反应但没有转换并给出错误：

Caused by: java.lang.IllegalArgumentException: The payload must be of type JobLaunchRequest. Object of class [java.io.File] must be an instance of class org.springframework.batch.integration.launch.JobLaunchRequest

如果存储桶中有两个文件（当应用程序启动时）FILE1.csv 和 FILE2.csv，则该作业会针对 FILE1.csv 正确运行，但会为 FILE2.csv 提供上述错误。

实现这样的事情的正确方法是什么？

明确地说，我想在这个存储桶中接收数千个 csv 文件，使用 Spring Batch 读取和处理，但我还需要尽快从 S3 获取每个新文件。

提前致谢。

解决方法

JobLaunchingGateway 确实只希望我们将 JobLaunchRequest 作为有效载荷。

既然你在 @InboundChannelAdapter(value = IN_CHANNEL_NAME,poller = @Poller(fixedDelay = "30")) bean 定义中有那个 S3InboundFileSynchronizingMessageSource，那么在 @ServiceActivator(inputChannel = IN_CHANNEL_NAME 之间没有 JobLaunchingGateway 转换器的情况下使用 FileMessageToJobRequest 真的是错误的.

您的 integrationFlow 对我来说看起来不错，但是您确实需要从 @InboundChannelAdapter bean 中删除那个 S3InboundFileSynchronizingMessageSource 并完全依赖于 c.poller() 配置。

另一种方法是离开 @InboundChannelAdapter，然后从 IntegrationFlow 开始 IN_CHANNEL_NAME，而不是 MessageSource。

由于您有多个轮询器针对同一个 S3 源，而且两者都基于相同的本地目录，因此看到这么多意外情况并不奇怪。

spring-batch spring-integration spring-integration-aws