Spring Batch4 监听 Job 之 JobExecutionListener

前面介绍了如何使用 Job，那么我们能否监听 Job 执行前后呢？答案是肯定的，Spring Batch4 提供了一个 JobExecutionListener 接口，该接口用来监听批处理作业（Job）的执行状态，如执行开始、执行结束，这就能让你在作业执行的前后阶段插入自定义逻辑。如下图：

注意，JobExecutionListener 接口可监听作业的开始和结束，接口中定义了两个核心方法：

• beforeJob(JobExecution jobExecution)：在作业启动之前调用，可用于初始化资源或者进行参数验证。

• afterJob(JobExecution jobExecution)：作业结束后调用，不管作业是正常完成、失败还是被中断，此方法都会被触发，适合用来释放资源或者记录审计信息。

JobExecutionListener 接口定义

JobExecutionListener 接口定义如下：

package org.springframework.batch.core;

/**
 * 在 {@link Job} 的生命周期中的特定点提供回调。
 * 如果实现者小心确保线程安全，或者为每个作业使用一个监听器实例
 *（假设作业实例本身不会被多个线程使用），则实现可以是有状态的。
 *
 * @author Dave Syer
 *
 */
public interface JobExecutionListener {

    /**
     * 作业执行前的回调。
     *
     * @param jobExecution the current {@link JobExecution}
     */
    void beforeJob(JobExecution jobExecution);

    /**
     * 作业完成后的回调。在成功和失败的执行之后都会调用。
     * 要针对特定状态执行逻辑，请使用“if (jobExecution.getStatus() == BatchStatus.X)”。
     *
     * @param jobExecution the current {@link JobExecution}
     */
    void afterJob(JobExecution jobExecution);

}

spring 广告位

简单示例

下面通过一个简单例子介绍如何使用 JobExecutionListener 监听 Job 的生命周期。

实现 JobExecutionListener 接口

创建 MyJobExecutionListener 类，实现 JobExecutionListener 接口。分别在 beforeJob() 和 afterJob() 方法中打印 Job 名称。代码如下：

package com.hxstrive.spring_batch.listenerDemo.listener;

import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionListener;

/**
 * JobExecutionListener 监听器：Job执行前后监听器
 * @author hxstrive.com
 */
public class MyJobExecutionListener implements JobExecutionListener {

    @Override
    public void beforeJob(JobExecution jobExecution) {
        System.out.println("MyJobExecutionListener beforeJob :: " + jobExecution.getJobInstance().getJobName());
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        System.out.println("MyJobExecutionListener afterJob :: " + jobExecution.getJobInstance().getJobName());
    }
}

创建 BatchConfig 配置

创建 BatchConfig 配置类，创建一个 Job，然后设置自定义的监听器 MyJobExecutionListener。代码如下：

package com.hxstrive.spring_batch.listenerJobDemo.config;

import com.hxstrive.spring_batch.listenerJobDemo.listener.MyJobExecutionListener;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Spring Batch 配置类
 * @author hxstrive.com
 */
@Configuration
public class BatchConfig {

    // 用于创建和配置 Job 对象的工厂类
    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    // 用于创建和配置 Step 对象的工厂类
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    // 创建 Step 对象
    @Bean
    public Step listenerJobDemoStep1() {
        return stepBuilderFactory.get("listenerJobDemoStep1")
                .tasklet(new Tasklet() {
                    @Override
                    public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception {
                        System.out.println(Thread.currentThread().getName() + " listenerJobDemoStep1");
                        return RepeatStatus.FINISHED; // 返回 FINISHED 表明任务执行结束
                    }
                }).build();
    }

    // 创建 Job 任务
    @Bean
    public Job listenerJobDemoJob() {
        return jobBuilderFactory.get("listenerJobDemo" + System.currentTimeMillis())
                .listener(new MyJobExecutionListener()) // 添加 Job 执行监听器
                .start(listenerJobDemoStep1())
                .build();
    }

}

上述代码中，使用 listener(new MyJobExecutionListener()) 代码为 Job 添加监听器。

运行&验证

运行应用程序，查看输出日志：

 注意，在 JobExecutionListener 接口的 beforeJob() 或 afterJob() 方法中，可以通过参数 JobExecution 的 getStatus() 方法获取 Job 执行状态，状态为 BatchStatus 枚举对象。

BatchStatus 用于表示批处理作业（Job）或步骤（Step）的执行状态，这个枚举类定义了作业生命周期中可能的各种状态，帮助你跟踪和管理批处理任务的执行情况。

BatchStatus 枚举类包含以下状态：

• COMPLETED：作业或步骤已成功完成。所有任务都按预期执行，没有错误发生。

• STARTING：作业或步骤正在启动过程中，尚未完全开始执行。

• STARTED：作业或步骤已成功启动，正在执行中。

• STOPPING：作业或步骤正在停止过程中，但尚未完全停止。

• STOPPED：作业或步骤已被成功停止，通常是通过外部干预（如手动停止）。

• FAILED：作业或步骤执行失败，可能是由于异常或错误导致的。

• ABANDONED：作业或步骤已被放弃，通常在处理失败的作业时，需要跳过某些已处理的项。

• UNKNOWN：状态未知，通常表示出现了意外情况或状态无法确定。

状态的简单用法如下：

@Override
public void afterJob(JobExecution jobExecution) {
    System.out.println("MyJobExecutionListener afterJob :: " + jobExecution.getJobInstance().getJobName());

    final BatchStatus status = jobExecution.getStatus();
    if(status.equals(BatchStatus.COMPLETED)) {
        // 作业执行成功
        System.out.println("Job completed successfully");
    } else if(status.equals(BatchStatus.FAILED)) {
        // 作业执行失败
        System.err.println("Job failed with exceptions: " + jobExecution.getAllFailureExceptions());
    }
}

作业的典型状态转换流程如下图：

BatchStatus 枚举能帮助你跟踪和管理批处理作业的执行状态。了解各种状态及其转换流程，能够更好地监控作业执行情况，并在必要时采取相应的措施（如重试失败的作业）。