Spring Batch中如何动态获取当前Job启动时间并修改Reader查询条件

核心思路 #

问题的关键在于：Step构建阶段Job尚未启动，无法获取当前启动时间；而JobListener的beforeJob虽能拿到时间，但此时Step已完成初始化。解决方案是利用Step执行阶段的回调机制，在Step启动后、Reader开始读取数据前，将当前Job启动时间传递给Reader。

方案一：让自定义Reader实现StepExecutionListener接口 #

让你的EtlUpdateReader直接实现StepExecutionListener，重写beforeStep方法——该方法会在Step启动后、Reader读取数据前执行，此时能拿到当前Job的启动时间，直接修改Reader的查询参数或SQL语句。

代码示例 #

public class EtlUpdateReader extends JdbcPagingItemReader<YourEntity> implements StepExecutionListener {

    @Override
    public void beforeStep(StepExecution stepExecution) {
        // 获取当前Job的启动时间
        Date currentJobStartTime = stepExecution.getJobExecution().getStartTime();
        // 将时间设置为查询参数（适用于带占位符的SQL，如WHERE update_time >= :startTime）
        this.setParameterValues(Collections.singletonMap("startTime", currentJobStartTime));
        
        // 如果需要动态拼接WHERE子句，可直接修改SQL
        // String originalSql = this.getSql();
        // String updatedSql = originalSql + " WHERE update_time >= ?";
        // this.setSql(updatedSql);
        // 注意：若用占位符，需同步调整参数设置逻辑
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return ExitStatus.COMPLETED;
    }

    // 原有Reader的初始化逻辑...
}

Step配置 #

在构建Step时，将Reader注册为Step的Listener：

@Bean
public Step etlStep(JobRepository jobRepository, PlatformTransactionManager transactionManager, EtlUpdateReader etlUpdateReader) {
    return new StepBuilder("etlStep", jobRepository)
            .<YourEntity, YourEntity>chunk(100, transactionManager)
            .reader(etlUpdateReader)
            .listener(etlUpdateReader) // 注册Listener
            .writer(yourEtlWriter())
            .build();
}

方案二：单独配置StepExecutionListener传递时间 #

如果不想修改Reader的原有代码，可单独实现一个StepExecutionListener，通过构造注入Reader实例，在beforeStep中传递启动时间。

代码示例 #

自定义Listener

public class JobStartTimeReaderListener implements StepExecutionListener {
    private final EtlUpdateReader etlUpdateReader;

    // 构造注入Reader
    public JobStartTimeReaderListener(EtlUpdateReader etlUpdateReader) {
        this.etlUpdateReader = etlUpdateReader;
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        Date startTime = stepExecution.getJobExecution().getStartTime();
        // 传递时间参数给Reader
        etlUpdateReader.setParameterValues(Collections.singletonMap("startTime", startTime));
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return ExitStatus.COMPLETED;
    }
}

Step配置

@Bean
public Step etlStep(JobRepository jobRepository, PlatformTransactionManager transactionManager, EtlUpdateReader etlUpdateReader) {
    return new StepBuilder("etlStep", jobRepository)
            .<YourEntity, YourEntity>chunk(100, transactionManager)
            .reader(etlUpdateReader)
            .listener(new JobStartTimeReaderListener(etlUpdateReader)) // 添加自定义Listener
            .writer(yourEtlWriter())
            .build();
}

方案三：通过JobExecutionListener+ItemStream传递时间 #

若你已实现JobExecutionListener，可在beforeJob中将启动时间存入JobExecution的ExecutionContext，再让Reader实现ItemStream接口，在open方法中读取时间参数。

代码示例 #

JobListener存储时间

public class JobStartTimeListener implements JobExecutionListener {
    @Override
    public void beforeJob(JobExecution jobExecution) {
        Date startTime = jobExecution.getStartTime();
        jobExecution.getExecutionContext().put("currentJobStartTime", startTime);
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        // 后续清理或日志逻辑
    }
}

Reader实现ItemStream读取时间

public class EtlUpdateReader extends JdbcPagingItemReader<YourEntity> implements ItemStream {

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        // 从ExecutionContext获取当前Job启动时间
        Date currentJobStartTime = (Date) executionContext.get("currentJobStartTime");
        this.setParameterValues(Collections.singletonMap("startTime", currentJobStartTime));
        // 调用父类的open方法完成初始化
        super.open(executionContext);
    }

    // 其他ItemStream方法可默认实现
    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {}

    @Override
    public void close() throws ItemStreamException {}

    // 原有Reader逻辑...
}

配置Job和Step

@Bean
public Job etlJob(JobRepository jobRepository, Step etlStep, JobStartTimeListener jobStartTimeListener) {
    return new JobBuilder("etlJob", jobRepository)
            .start(etlStep)
            .listener(jobStartTimeListener) // 注册JobListener
            .build();
}

关键说明 #

以上方案的核心是利用Step运行时的回调时机：无论是StepExecutionListener的beforeStep，还是ItemStream的open方法，都是在Step启动后、Reader开始读取数据前执行，此时Job已完成启动，能获取到当前Job的真实启动时间，而非上次运行的时间。

内容的提问来源于stack exchange，提问作者maw2be