超时后重新启动步骤(或作业) [英] Restart step (or job) after timeout occours

问题描述

是否可以在超时时自动重新启动作业或步骤？ 我尝试了重试和跳过(跳过，因为如果没有发生错误，作业每30分钟重新运行一次)步骤，如下所示：

<step id="jobTest.step1">
  <tasklet>
    <transaction-attributes timeout="120"/>
    <chunk reader="testReader" processor="testProcessor" writer="testWriter" commit-interval="10"  retry-limit="3" >
      <retryable-exception-classes>
        <include class="org.springframework.transaction.TransactionTimedOutException"/>
       </retryable-exception-classes>
    </chunk>
    <listeners>
      <listener ref="stepListener" />
    </listeners>
  </tasklet>
</step>

我也尝试了跳过策略，但没有得到满意的结果。 发生超时时，我只需要重新启动此步骤(或整个作业)。

更新

我也试过了，但没有成功： Spring batch: Retry job if does not complete in particular time

推荐答案

重试/跳过功能适用于面向块的容错步骤中块内的项目，而不适用于步骤级或作业级。您的需求中实际上有两个截然不同的东西：

1.如何在给定超时后停止作业？

除了在超时后外部调用JobOperator#stop之外，您还可以通过StepExecution#isTerminateOnly标志发送停止信号，从作业本身停止作业。其想法是能够访问步骤执行，以便在特定超时后设置该标志。这取决于步骤的微线程类型：

简单Tasklet

对于简单的微线程，您可以通过ChunkContext访问步骤执行。下面是一个示例：

import java.time.Duration;
import java.util.Date;

import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;

public class MyTasklet implements Tasklet {

    private static final int TIMEOUT = 120; // in minutes (can be turned into a configurable field through a constructor)

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        if (timeout(chunkContext)) {
            chunkContext.getStepContext().getStepExecution().setTerminateOnly();
        }
        // do some work
        if (moreWork()) {
            return RepeatStatus.CONTINUABLE;
        } else {
            return RepeatStatus.FINISHED;
        }
    }

    private boolean timeout(ChunkContext chunkContext) {
        Date startTime = chunkContext.getStepContext().getStepExecution().getJobExecution().getStartTime();
        Date now = new Date();
        return Duration.between(startTime.toInstant(), now.toInstant()).toMinutes() > TIMEOUT;
    }

    private boolean moreWork() {
        return false; // TODO implement logic
    }
}

此微程将定期检查是否超时，并相应地停止该步骤(以及周围的作业)。

面向块的微线程

在这种情况下，您可以使用步骤侦听器并在生命周期方法之一(afterRead、afterWrite等)中设置terminateOnly标志。下面是一个示例：

import java.time.Duration;
import java.util.Date;

import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.listener.StepListenerSupport;
import org.springframework.batch.core.scope.context.ChunkContext;

public class StopListener extends StepListenerSupport {

    private static final int TIMEOUT = 120; // in minutes (can be made configurable through constructor)

    private StepExecution stepExecution;

    @Override
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public void afterChunk(ChunkContext context) { // or afterRead, or afterWrite, etc.
        if (timeout(context)) {
            this.stepExecution.setTerminateOnly();
        }
    }

    private boolean timeout(ChunkContext chunkContext) {
        Date startTime = chunkContext.getStepContext().getStepExecution().getJobExecution().getStartTime();
        Date now = new Date();
        return Duration.between(startTime.toInstant(), now.toInstant()).toMinutes() > TIMEOUT;
    }
}

思路相同，您需要定期检查时间并在适当的时候设置标志。

这两种方式都会使您的作业处于STOPPED状态，即可重新启动状态。批处理作业过去在批处理窗口中执行，一个常见的要求是在窗口关闭时(优雅地)停止它们。前面的技术是可行的。

Spring batch: Retry job if does not complete in particular time中的答案不是一个好的IMO选项，因为它会突然终止当前块的事务，并使作业处于FAILED状态(这也是一个可重新启动的状态)。但是，通过查看处于FAILED状态的作业，无法区分真正的失败和故意停止。考虑到故意希望作业在批处理窗口结束时停止的要求，我认为应该在下一个窗口中正常停止并重新启动该作业。

2.超时后如何自动重启作业？

现在您知道了如何在超时后停止作业，您可以在作业启动器周围使用RetryTemplate，并在适当的时候重新启动作业。下面是一个示例：

public static void main(String[] args) throws Throwable {
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3));

    ApplicationContext applicationContext = new AnnotationConfigApplicationContext(MyJob.class);
    JobLauncher jobLauncher = applicationContext.getBean(JobLauncher.class);
    Job job = applicationContext.getBean(Job.class);
    JobParameters jobParameters = new JobParametersBuilder()
            .addDate("runtime", new Date())
            .toJobParameters();

    retryTemplate.execute((RetryCallback<JobExecution, Throwable>) retryContext -> {
        JobExecution jobExecution = jobLauncher.run(job, jobParameters);
        if (jobExecution.getExitStatus().getExitCode().equals(ExitStatus.STOPPED.getExitCode())) {
            throw new Exception("Job timeout");
        }
        return jobExecution;
    });
}

如果作业以STOPPED状态完成(例如，由于如前所述的超时)，则该作业最多将自动重新运行3次。

希望这对您有帮助。

这篇关于超时后重新启动步骤(或作业)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！