如何将JobStep子作业的Execution Context值提升至父作业？

问题背景 #

我基于Spring Batch开发了一个报表运行框架，流程分三步：

• 用Tasklet从数据库获取报表定义（含作业名、静态参数、接收用户列表），存入作业执行上下文；
• 基于上下文里的报表定义创建JobStep，执行报表生成作业并产出文件；
• 用Tasklet把生成的文件分发给指定用户。

现在的需求是：报表生成作业（子作业）把生成的文件名存入自身Execution Context后，要传递到父作业的Execution Context供第三步使用。但发现子作业的JobExecution仅在doExecute方法内有效，只用来设置步骤退出状态，找不到现成的传递方式。考虑过子类化JobStep，实现类似ExecutionContextPromotionListener的功能来提升子作业的值到父步骤，想确认有没有更合适的方法。

当前JobStep定义代码 #

@JobScope
@Bean(name="runReportStep")
Step buildRunReportStep(@Value("#{jobExecutionContext['reportDefn']}") ReportDefinition rptDfn) {
    try {
        Job job = jobRegistry.getJob(rptDfn.getJobName());
        return new StepBuilder("runReportStep", jobRepository)
                .job(job)
                .launcher(jobLauncher)
                .build();
    } catch (NoSuchJobException e) {
        return null;
    }
}

Spring Batch 5.1.0中JobStep的doExecute方法源码 #

/**
 * Execute the job provided by delegating to the {@link JobLauncher} to prevent
 * duplicate executions. The job parameters will be generated by the
 * {@link JobParametersExtractor} provided (if any), otherwise empty. On a restart,
 * the job parameters will be the same as the last (failed) execution.
 *
 * @see AbstractStep#doExecute(StepExecution)
 */
@Override
protected void doExecute(StepExecution stepExecution) throws Exception {

    ExecutionContext executionContext = stepExecution.getExecutionContext();

    executionContext.put(STEP_TYPE_KEY, this.getClass().getName());

    JobParameters jobParameters;
    if (executionContext.containsKey(JOB_PARAMETERS_KEY)) {
        jobParameters = (JobParameters) executionContext.get(JOB_PARAMETERS_KEY);
    }
    else {
        jobParameters = jobParametersExtractor.getJobParameters(job, stepExecution);
        executionContext.put(JOB_PARAMETERS_KEY, jobParameters);
    }

    JobExecution jobExecution = jobLauncher.run(job, jobParameters);

    stepExecution.setExitStatus(determineStepExitStatus(stepExecution, jobExecution));

    if (jobExecution.getStatus().isUnsuccessful()) {
        // AbstractStep will take care of the step execution status
        throw new UnexpectedJobExecutionException("Step failure: the delegate Job failed in JobStep.");
    }
    else if (jobExecution.getStatus().equals(BatchStatus.STOPPED)) {
        stepExecution.setStatus(BatchStatus.STOPPED);
    }
}

解决方案 #

方案一：自定义JobStep实现上下文值自动提升 #

你考虑的子类化JobStep方案完全可行，而且是最贴合Spring Batch上下文传递逻辑的实现方式。核心思路是在子作业执行完成后，把它上下文里的指定key复制到父步骤的Execution Context，再通过ExecutionContextPromotionListener把这些key提升到父作业的上下文。

具体实现步骤：

1. 自定义PromotingJobStep继承自JobStep，重写doExecute方法，添加上下文复制逻辑：

public class PromotingJobStep extends JobStep {

    private List<String> keysToPromote;

    public PromotingJobStep(Builder builder) {
        super(builder);
    }

    @Override
    protected void doExecute(StepExecution stepExecution) throws Exception {
        ExecutionContext executionContext = stepExecution.getExecutionContext();
        executionContext.put(STEP_TYPE_KEY, this.getClass().getName());

        JobParameters jobParameters;
        if (executionContext.containsKey(JOB_PARAMETERS_KEY)) {
            jobParameters = (JobParameters) executionContext.get(JOB_PARAMETERS_KEY);
        } else {
            jobParameters = getJobParametersExtractor().getJobParameters(getJob(), stepExecution);
            executionContext.put(JOB_PARAMETERS_KEY, jobParameters);
        }

        // 执行子作业
        JobExecution jobExecution = getJobLauncher().run(getJob(), jobParameters);
        stepExecution.setExitStatus(determineStepExitStatus(stepExecution, jobExecution));

        // 新增：将子作业上下文的指定key复制到父步骤上下文
        if (keysToPromote != null && !keysToPromote.isEmpty()) {
            ExecutionContext childContext = jobExecution.getExecutionContext();
            ExecutionContext parentStepContext = stepExecution.getExecutionContext();
            keysToPromote.forEach(key -> {
                if (childContext.containsKey(key)) {
                    parentStepContext.put(key, childContext.get(key));
                }
            });
        }

        // 原有错误处理逻辑
        if (jobExecution.getStatus().isUnsuccessful()) {
            throw new UnexpectedJobExecutionException("Step failure: the delegate Job failed in JobStep.");
        } else if (jobExecution.getStatus().equals(BatchStatus.STOPPED)) {
            stepExecution.setStatus(BatchStatus.STOPPED);
        }
    }

    public void setKeysToPromote(List<String> keysToPromote) {
        this.keysToPromote = keysToPromote;
    }
}

2. 在创建JobStep时使用自定义类，指定需要提升的key：

@JobScope
@Bean(name="runReportStep")
Step buildRunReportStep(@Value("#{jobExecutionContext['reportDefn']}") ReportDefinition rptDfn) {
    try {
        Job job = jobRegistry.getJob(rptDfn.getJobName());
        StepBuilder builder = new StepBuilder("runReportStep", jobRepository);
        PromotingJobStep jobStep = new PromotingJobStep(builder.job(job).launcher(jobLauncher).getBuilder());
        // 指定要从子作业上下文复制的key，比如生成的文件名对应的key
        jobStep.setKeysToPromote(Collections.singletonList("generatedFileName"));
        return jobStep;
    } catch (NoSuchJobException e) {
        return null;
    }
}

3. 配置监听器，把父步骤上下文的key提升到父作业上下文：

@Bean
public ExecutionContextPromotionListener reportPromotionListener() {
    ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
    // 设置要提升到作业上下文的key
    listener.setKeys(new String[]{"generatedFileName"});
    listener.setStrict(false); // 允许key不存在时不抛出异常
    return listener;
}

// 在父作业中给runReportStep绑定监听器
@Bean
public Job reportJob(Step fetchReportDefnStep, Step runReportStep, Step distributeFileStep) {
    return new JobBuilder("reportJob", jobRepository)
            .start(fetchReportDefnStep)
            .next(runReportStep)
            .listener(reportPromotionListener())
            .next(distributeFileStep)
            .build();
}

方案二：通过JobRepository查询子作业上下文 #

如果不想自定义JobStep，可以在第三步的分发Tasklet中，通过父作业上下文获取子作业的JobParameters，再用JobRepository查询子作业的JobExecution，进而拿到生成的文件名。这种方式无需修改核心组件，但逻辑相对繁琐。

示例实现：

@Component
public class DistributeFileTasklet implements Tasklet {

    private final JobRepository jobRepository;
    private final JobRegistry jobRegistry;

    public DistributeFileTasklet(JobRepository jobRepository, JobRegistry jobRegistry) {
        this.jobRepository = jobRepository;
        this.jobRegistry = jobRegistry;
    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        // 获取父步骤上下文里存储的子作业JobParameters（原JobStep会自动存入JOB_PARAMETERS_KEY）
        JobParameters childJobParams = (JobParameters) chunkContext.getStepContext()
                .getStepExecutionContext()
                .get(JobStep.JOB_PARAMETERS_KEY);
        
        // 根据子作业名和参数查询对应的JobExecution
        Job childJob = jobRegistry.getJob(rptDfn.getJobName()); // 可从reportDefn中获取作业名
        List<JobExecution> childExecutions = jobRepository.findJobExecutions(childJob, childJobParams);
        
        if (!childExecutions.isEmpty()) {
            // 取最新执行的子作业实例
            JobExecution childJobExecution = childExecutions.get(childExecutions.size() - 1);
            String generatedFileName = (String) childJobExecution.getExecutionContext().get("generatedFileName");
            // 执行文件分发逻辑，比如发送邮件给接收用户列表
        }
        
        return RepeatStatus.FINISHED;
    }
}

总结 #

• 方案一（自定义JobStep）逻辑清晰，符合Spring Batch的设计思想，是更推荐的实现方式；
• 方案二（JobRepository查询）无需修改核心组件，但依赖JobRepository的查询能力，适合不想自定义组件的场景。

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