如果开始运行同一个作业（Spring批处理），则运行作业将停止

Question

我有点困惑，因为当通过HTTP请求启动Spring Batch作业的执行时，如果我在作业执行时收到另一个HTTP请求来启动相同的作业，但参数不同，则正在执行的作业停止未完成并开始处理新作业。

我开发了一个API REST来加载和处理Excel文件的内容。web服务公开了两个endpoint，一个用于加载、验证和存储数据库中Excel文件的内容，另一个用于开始处理存储在html" target="_blank">数据库中的记录。

>

POST /api/excel/upload此endpoint接收Excel文件。当接收到请求时，每个文件都会被分配一个唯一标识符并验证其内容。如果内容正确，它会将其插入到等待处理的临时表中。

获取/api/Excel/process？id=x此endpoint接收要处理的文件的标识符。当收到请求时，一个Spring批处理作业开始处理临时表中的记录。

• 控制器

@PostMapping(produces = {APPLICATION_JSON_VALUE})
public ResponseEntity<Page<ExcelLoad>> post(@RequestParam("file") MultipartFile multipartFile)
{
    return super.getResponse().returnPage(service.upload(multipartFile));
}

@GetMapping(value = "/process", produces = APPLICATION_JSON_VALUE)
public DeferredResult<ResponseEntity<Void>> get(@RequestParam("id") Integer idCarga)
{
    DeferredResult<ResponseEntity<Void>> response = new DeferredResult<>(1000L);
    response.onTimeout(() -> response.setResult(super.getResponse().returnVoid()));

    ForkJoinPool.commonPool().submit(() -> service.startJob(idCarga));

    return response;
}

我使用DelferredResult在收到请求后向客户端发送响应，而不用等待作业完成

• 服务

public void startJob(int idCarga)
{
    JobParameters params = new JobParametersBuilder()
            .addString("mainJob", String.valueOf(System.currentTimeMillis()))
            .addString("idCarga", String.valueOf(idCarga))
            .toJobParameters();

    try
    {
        jobLauncher.run(job, params);
    }
    catch (JobExecutionException e)
    {
        log.error("---ERROR: {}", e.getMessage());
    }
}

• 一批

@Bean
public Step mainStep(ReaderImpl reader, ProcessorImpl processor, WriterImpl writer)
{
    return stepBuilderFactory.get("step")
            .<List<ExcelLoad>, Invoice>chunk(10)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .faultTolerant().skipPolicy(new ExceptionSkipPolicy())
            .listener(stepSkipListener)
            .build();
}

@Bean
public Job mainJob(Step mainStep)
{
    return jobBuilderFactory.get("mainJob")
                            .listener(mainJobExecutionListener)
                            .incrementer(new RunIdIncrementer())
                            .start(mainStep)
                            .build();
}

在执行一些测试时，我观察到以下行为：

>

• 记录处理文件1:3606（预期为3606）。
• 记录处理文件2:1776（预期为1776）。

如果我向endpoint/进程发出请求，请求第一个处理文件1，在它完成之前，我再次请求处理文件2：在这种情况下，并不是所有存储在临时表中的记录都被处理：

• 记录处理文件1:1080（预计3606）
• 记录处理文件2:1774（预计为1776）

Answer 1

多亏了@Mahmoud Ben Hassine的帮助，我才得以解决这个问题。为了帮助实现，如果有人遇到这个问题，我将分享代码，在我的例子中，这些代码已经解决了这个问题：

• 控制器

@Autowired
private JobLauncher jobLauncher;

@Autowired
private Job job;

@GetMapping(value = "/process", produces = APPLICATION_JSON_VALUE)
public void get(@RequestParam("id") Integer idCarga) throws JobExecutionException
{
    JobParameters params = new JobParametersBuilder()
            .addString("mainJob", String.valueOf(System.currentTimeMillis()))
            .addString("idCarga", String.valueOf(idCarga))
            .toJobParameters();

    jobLauncher.run(job, params);
}

• 批处理配置、作业和步骤

@Configuration
@EnableBatchProcessing
public class BatchConfig extends DefaultBatchConfigurer
{
    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private StepSkipListener stepSkipListener;

    @Autowired
    private MainJobExecutionListener mainJobExecutionListener;

    @Bean
    public TaskExecutor taskExecutor()
    {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setMaxPoolSize(10);
        taskExecutor.setThreadNamePrefix("batch-thread-");

        return taskExecutor;
    }

    @Bean
    public JobLauncher jobLauncher() throws Exception
    {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(getJobRepository());
        jobLauncher.setTaskExecutor(taskExecutor());
        jobLauncher.afterPropertiesSet();

        return jobLauncher;
    }

    @Bean
    public Step mainStep(ReaderImpl reader, ProcessorImpl processor, WriterImpl writer)
    {
        return stepBuilderFactory.get("step")
                .<List<ExcelLoad>, Invoice>chunk(10)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .faultTolerant().skipPolicy(new ExceptionSkipPolicy())
                .listener(stepSkipListener)
                .build();
    }

    @Bean
    public Job mainJob(Step mainStep)
    {
        return jobBuilderFactory.get("mainJob")
                                .listener(mainJobExecutionListener)
                                .incrementer(new RunIdIncrementer())
                                .start(mainStep)
                                .build();
    }
}

如果在应用了这个代码之后，就像我遇到的那样，你在数据库中插入记录时也遇到了问题，你可以通过这个问题，我也把适用于我的代码放在这里。

Answer 2

JobLauncher不会停止作业执行，它只会启动它们。Spring Batch提供的默认作业启动程序是SimpleZoblancher，它将作业启动委托给TaskExecutor。现在，根据您使用的任务执行器实现以及如何配置它以启动并发任务，您可以看到不同的行为。例如，当启动新作业执行并将新任务提交给任务执行者时，如果所有工作人员都很忙，任务执行者可以决定拒绝此提交，或者将其放入等待队列，或者停止另一个任务并提交新任务。这些策略取决于几个参数（TaskExecutor实现、幕后使用的队列类型、RejectedExecutionHandler实现等）。

在您的情况下，您似乎在使用以下内容：

ForkJoinPool.commonPool().submit(() -> service.startJob(idCarga));

因此，您需要检查这个池在处理新任务提交方面的行为（我猜这就是停止工作的原因，但您需要确认这一点）。也就是说，我不明白你为什么需要这个。如果您的要求如下：

我使用DelferredResult在收到请求后向客户端发送响应，而不用等待作业完成

然后，可以在作业启动器中使用异步任务执行器实现（如线程池任务执行器），请参阅从Web容器中运行作业。