我正在关注文档:Spring Batch Integration与Integration AWS相结合,用于合并AWS S3。
但是在某些情况下,每个文件的批处理执行不起作用。
AWS S3 Pooling正常运行,因此当我放置新文件或启动应用程序且存储桶中有文件时,应用程序将与本地目录同步:
@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(应用程序启动时存储桶为空):
情况2(应用程序启动时,存储桶中已经有一个文件):
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]
情况3(存储桶有多个文件):
因此,按照文档,我将网关更改为:
@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"))
在S3InboundFileSynchronizingMessageSource
bean定义上有了它,那么@ServiceActivator(inputChannel = IN_CHANNEL_NAME
在JobLaunchingGateway
没有FileMessageToJobRequest
变压器的情况下拥有它确实是错误的。
您integrationFlow
看起来对我来说还可以,但是您确实需要@InboundChannelAdapter
从S3InboundFileSynchronizingMessageSource
Bean中删除它,并完全依赖于c.poller()
配置。
另一种方法是留下@InboundChannelAdapter
,但随后开始IntegrationFlow
从IN_CHANNEL_NAME
不是一个MessageSource
。
由于您具有针对同一S3源的多个轮询器,而且这两个轮询器均基于同一本地目录,因此看到如此多的意外情况就不足为奇了。
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句