Spring Batch - 并行运行多个作业

Question

我是 Spring Batch 的新手，不知道如何做到这一点..

基本上，我有一个 spring 文件轮询器，它每 N 分钟运行一次，以在某个目录中查找具有某些名称(例如:A.txt 和 B.txt)的文件。在任何时刻，该目录中最多可以有 2 个文件(A 和 B)。通过Spring Batch Job，这两个文件将被处理并持久化到2个不同的数据库表中。

这些文件有些相似，因此使用相同的处理器/编写器。

现在按照我设置的方式，每个轮询周期 1 个文件都会被拾取并运行作业。

假设目录中有 2 个文件(A.txt 和 B.txt)，有没有办法创建 2 个作业，以便两个作业可以并行运行？

Answer 1

有非常好的方法可以使用 Spring 以异步模式运行作业，这只是如何配置 JobLauncher 的问题。 JobLauncher 有一个 taskExecutor 属性，并且可以根据分配给该属性的实现来激活异步执行。

您可以找到 Spring 可以提供的所有 TaskExecutor 类型，并根据您的需求选择完成批处理异步作业的最佳方法。 Task Executors Types in Spring

例如，SimpleAsyncTaskExecutor 是一个任务执行器，它将在任何调用上创建一个新的Thread，如果执行频率很高，则可能会产生性能问题。另一方面，还有一些 TaskExecutors 类型提供池化功能，以便重用资源并最大限度地提高系统效率。

这是一个如何配置ThreadPoolTask​​Executor的小示例:

A) 配置ThreadPoolTask​​Executor Bean

@Bean
    public ThreadPoolTaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(15);
    taskExecutor.setMaxPoolSize(20);
    taskExecutor.setQueueCapacity(30);
    return taskExecutor;
}

B) 配置 JobLauncher Bean

   @Bean
    public JobLauncher jobLauncher(ThreadPoolTaskExecutor taskExecutor, JobRepository jobRepository){
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setTaskExecutor(taskExecutor);
        jobLauncher.setJobRepository(jobRepository);
        return jobLauncher;
    }

C) 注入(inject)您的 JobLauncher 和 Jobs 配置

@Autowired
private JobLauncher jobLauncher;

@Autowired
@Qualifier("job1-file-A")
private Job job1;

@Autowired
@Qualifier("job2-file-B")
private Job job2;

D)安排作业

@Scheduled(cron = "*/1 * * * * *")
public void run1(){
    Map<String, JobParameter> confMap = new HashMap<>();
    confMap.put("time", new JobParameter(System.currentTimeMillis()));
    JobParameters jobParameters = new JobParameters(confMap);
    try {
        jobLauncher.run(job1, jobParameters);
    }catch (Exception ex){
        logger.error(ex.getMessage());
    }

}

@Scheduled(cron = "*/1 * * * * *")
public void run2(){
    Map<String, JobParameter> confMap = new HashMap<>();
    confMap.put("time", new JobParameter(System.currentTimeMillis()));
    JobParameters jobParameters = new JobParameters(confMap);
    try {
        jobLauncher.run(job2, jobParameters);
    }catch (Exception ex){
        logger.error(ex.getMessage());
    }

}

E) 最后在 SpringBoot 类上 @EnableBatchProcessing 和 @EnableScheduling

@EnableBatchProcessing
@EnableScheduling
@SpringBootApplication
public class MyBatchApp {