将 Spring Batch Tasklet 失败消息传递给报告步骤。

Question

我正在使用带有 OpenCSV 的 Spring Batch Tasklet 来读取我的 CSV 文件。在问这个问题之前，我知道块，但是在后面的步骤中文件之间存在交叉验证，所以我必须继续使用Tasklet。我要做的是向我的报告步骤报告丢失的文件或解析错误。我不确定向下一步报告失败的正确方法应该是什么。我有以下代码。读取文件的初始步骤。public class CsvBatchReader<T> implements Tasklet, StepExecutionListener {    private final Logger logger = LoggerFactory.getLogger(CsvBatchReader.class);    private List batch;    private final Class<T> clazz;    private Path path;    public CsvBatchReader(Class<T> clazz, Path path) {        this.clazz = clazz;        this.path = path;    }    @Override    public void beforeStep(StepExecution stepExecution) {        logger.info("Reader initialized - " + clazz.getSimpleName());        batch = new ArrayList();    }    @Override    public ExitStatus afterStep(StepExecution stepExecution) {        logger.info("Reader ended - " + clazz.getSimpleName());        return ExitStatus.COMPLETED;    }    @Override    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws UnexpectedJobExecutionException {        logger.info("Reader execute - " + clazz.getSimpleName());        ICsvToBean csvToBean = new CsvToBean(clazz, path);        try {            batch = csvToBean.readCsv();        } catch(IOException ex) {            // error message being caught from my csvToBean class.             throw new UnexpectedJobExecutionException("Invalid file " + ex.getMessage());        }        return RepeatStatus.FINISHED;    }}报告步骤我不确定如何传入异常消息，或者是否定义了在不使用 Step Execution Context 的情况下传入失败消息的方法。public class CsvBatchReporting implements Tasklet, StepExecutionListener {    private final Logger logger = LoggerFactory.getLogger(CsvBatchCrossValidation.class);    private List errorMessages;    private List skippedInserts;

Answer 1

我不确定如何传入异常消息，或者是否定义了在不使用 Step Execution Context 的情况下传入失败消息的方法。

您可以从作业执行中访问上一步中抛出的异常。下面是一个例子：

import java.util.List;

import org.springframework.batch.core.Job;

import org.springframework.batch.core.JobExecution;

import org.springframework.batch.core.JobParameters;

import org.springframework.batch.core.Step;

import org.springframework.batch.core.StepExecution;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;

import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;

import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;

import org.springframework.batch.core.launch.JobLauncher;

import org.springframework.batch.repeat.RepeatStatus;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.ApplicationContext;

import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

@EnableBatchProcessing

public class MyJob {

    @Autowired

    private JobBuilderFactory jobs;

    @Autowired

    private StepBuilderFactory steps;

    @Bean

    public Step step1() {

        return steps.get("step1")

                .tasklet((contribution, chunkContext) -> {

                    System.out.println("hello");

                    throw new Exception("Boom!");

                })

                .build();

    }

    @Bean

    public Step step2() {

        return steps.get("step2")

                .tasklet((contribution, chunkContext) -> {

                    JobExecution jobExecution = chunkContext.getStepContext().getStepExecution().getJobExecution();

                    StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); // TODO properly get the stepExecution of the previous step

                    List<Throwable> failureExceptions = stepExecution.getFailureExceptions();

                    if (!failureExceptions.isEmpty()) {

                        Throwable throwable = failureExceptions.get(0);

                        System.out.println("Looks like step1 has thrown an exception: " + throwable.getMessage());

                    }

                    System.out.println("world");

                    return RepeatStatus.FINISHED;

                })

                .build();

    }

    @Bean

    public Job job() {

        return jobs.get("job")

                    .flow(step1())

                    .on("*").to(step2())

                    .build()

                .build();

    }

    public static void main(String[] args) throws Exception {

        ApplicationContext context = new AnnotationConfigApplicationContext(MyJob.class);

        JobLauncher jobLauncher = context.getBean(JobLauncher.class);

        Job job = context.getBean(Job.class);

        jobLauncher.run(job, new JobParameters());

    }

}

此示例打印：

hello

Looks like step1 has thrown an exception: Boom!

world

显然，您需要确保在所有情况下 step1 都流向 step2（因此是流定义）。

希望这可以帮助。