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

3 回答

• 3

  我不确定如何传递异常消息，或者是否有定义的方法来传递失败消息而不使用步骤执行上下文 .

  您可以访问作业执行的上一步中抛出的异常 . 这是一个例子：

  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（因此流程定义） .

  希望这可以帮助 .

  回复于 2024-05-11T11:19:22+08:00
• 1

  考虑使 errorMessages 成为一个bean：

  // somewhere convenient...
  @Bean
  public List<String> errorMessages() {
      return new ArrayList<>();
  }
  

  并将 errorMessages 注入两个tasklet .

  在 CsvBatchReader tasklet中，检查 execute(...) 抛出的任何异常，并根据需要更新 errorMessages ：

  public class CsvBatchReader<T> implements Tasklet, StepExecutionListener {
  
      @Override
      public ExitStatus afterStep(StepExecution stepExecution) {
          List<Throwable> failures = stepExecution.getFailureExceptions();
          if (!failures.isEmpty())  {
              errorMessages.add(...);
          }
  
          logger.info("Reader ended - " + clazz.getSimpleName());
          return ExitStatus.COMPLETED;  // return INVALID if !failures.isEmpty() ?
      }
  }
  

  自 errorMessages 被注入 CsvBatchReporting ，然后：

  public class CsvBatchReporting implements Tasklet, StepExecutionListener {
  
  @Override
  public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
      logger.info("Reporting execute");
  
      //Email Error 
      String body = null;
      String subject = "job finished: ";
      if (!errorMessages.isEmpty())  {
         subject += "ERROR";
         body = ... // from errorMessages
      } else {
         subject += "SUCCESS";
      }
  
      return RepeatStatus.FINISHED;
  }
  
  }
  

  回复于 2024-05-11T11:19:22+08:00
• 0

  虽然我原来的问题是关于将异常从一步传递到下一步，但我想指出一种使用 exitStatus 的替代方法 .

  在我的阅读器步骤中，我可以捕获我的csvToBean异常并创建一个类似的存在状态 .

  @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) {
          chunkContext.getStepContext().getStepExecution().setExitStatus(new ExitStatus("FAILED", ex.getMessage()));
      }
  
      return RepeatStatus.FINISHED;
  }
  

  然后我会在我的工作中创建条件流程 .

  @Bean
  public Job primaryCareJob(@Qualifier("reportingStep") Step reportingStep, @Qualifier("crossValidationStep") Step crossValidationStep) {
      logger.info("Start PrimaryCare Job");
  
      return jobs.get("primaryCareJob")
              .start(readPrimaryCareStep()).on("FAILED").to(reportingStep)
              .from(readPrimaryCareStep()).on("*").to(readPrimaryCareDetailStep())
  
              .from(readPrimaryCareDetailStep()).on("FAILED").to(reportingStep)
              .from(readPrimaryCareDetailStep()).on("*").to(processPrimaryCareStep())
  
              .from(processPrimaryCareStep()).on("INVALID").to(reportingStep)
              .from(processPrimaryCareStep()).on("*").to(processPrimaryCareDetailStep())
  
              .from(processPrimaryCareDetailStep()).on("INVALID").to(reportingStep)
              //Other steps
  
              .from(reportingStep).on("*").end()
              .from(reportingStep).on("*").fail()
  
              .build()
          .build();
  }
  

  最后，在我的阅读器tasklet步骤中，我会像这样检索existStatus

  @Override
  public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
      JobExecution jobExecution = chunkContext.getStepContext().getStepExecution().getJobExecution();
  
      Collection<StepExecution> stepExecutions = jobExecution.getStepExecutions();
  
      Map<String, String> result = stepExecutions.stream()
          .filter(x -> x.getExitStatus().getExitCode().equals("FAILED"))
          .collect(Collectors.toMap(StepExecution::getStepName, x -> x.getExitStatus().getExitDescription()));
  
      result.forEach((k, v) -> {
          System.out.println("Step " + k + " failure " + v);
      });
  }
  

  回复于 2024-05-11T11:19:22+08:00