Spring Batch 3.0：StepExecutionListener 用于分区 Step 并将执行上下文值级联到分区作业

Question

给定一个 Spring Batch 使用分区的作业：

<job id="reportingJob" xmlns="http://www.springframework.org/schema/batch">
        <batch:listeners>
            <batch:listener ref="reportingJobExecutionListenerr" />
        </batch:listeners>
        <batch:step id="reportingMasterStep">
            <partition step="reportingSlaveStep"
                partitioner="reportingPartitioner">
                <batch:handler grid-size="10" task-executor="taskExecutor" />
            </partition>
        </batch:step>
</job>

而reportingSlaveStep 定义为：

<step id="reportingSlaveStep" xmlns="http://www.springframework.org/schema/batch">
        <job ref="reportingSlaveJob" />
</step>

而reportingSlaveJob 定义为：

    <job id="reportingSlaveJob" xmlns="http://www.springframework.org/schema/batch">
        <batch:listeners>
            <batch:listener ref="reportsOutputListener" />
        </batch:listeners>
        <batch:split id="reportsCreationSplit"
            task-executor="taskExecutor">
            <batch:flow>
                <batch:step id="basicReportStep">
                    <tasklet throttle-limit="5" task-executor="taskExecutor">
                        <batch:chunk reader="basicReportReader"
                            writer="basicReportWriter" commit-interval="500" />
                    </tasklet>
                </batch:step>
            </batch:flow>
            <batch:flow>
                <batch:step id="advancedReportStep">
                    <tasklet throttle-limit="5" task-executor="taskExecutor">
                        <batch:chunk reader="advancedReportDataReader" writer="advancedReportWriter"
                            commit-interval="500" />
                    </tasklet>
                </batch:step>
            </batch:flow>
       </batch:split>
     </job>

我现在有 2 个问题：

1. 我想为每个分区创建一个新的reportsOutputListener 实例。我可以通过将 reportsOutputListener 设为 Step 范围 bean 来实现这一点吗？
2. 我希望能够访问为reportingJob 创建的相同jobExecutionContext，以便在reportingSlaveJob 中访问。我是否需要对此进行任何特殊处理，或者reportingJob 中的jobExecutionContext 实例是否也被reportingSlaveStepSlaveJob 使用？
3. 编辑：当我运行上述作业时，有时我会收到一个异常，说“该作业的作业执行已经在运行”，而其他时候我在MapExecutionContextDao.java:130 上得到一个NullPointerException。

EDIT ：另请注意，对于第 2 点，slaveJob 无法访问 @987654340 在 stepExecutionContext 中添加的值（在 spring config xml 中使用 #{stepExecutionContext['msbfBatchId']} 访问） @。 stepExecutionContext 中针对密钥的值显示为 null。

Answer 1

我想为每个创建一个新的 reportsOutputListener 实例 分割。我可以通过将 reportsOutputListener 设为 Step 来实现吗 作用域bean？

答案是是的。 （正如 Mahmoud Ben Hassine 在 cmets 中提到的）

我希望能够访问为 在 reportingSlaveJob 中可访问的 reportingJob。我需要任何 对此的特殊处理或者是同一个jobExecutionContext实例 在 reportingJob 中也被 reportingSlaveStepSlaveJob 使用？

答案是否。我深入研究了 Spring Batch 代码，发现JobStep 使用JobParametersExtractor 将值从stepExecutionContext 复制到JobParameters。这意味着reportingSlaveJob 可以从JobParameters 而不是StepExecutionContext 访问这些值。也就是说，出于某种原因，Srping Batch 3.0 中的 DefaultJobParametersExtractor 实现似乎没有像预期的那样将值复制到 jobParameters。我最终编写了以下自定义提取器：

public class CustomJobParametersExtractor implements JobParametersExtractor {

    private Set<String> keys;

    public CustomJobParametersExtractor () {
        this.keys = new HashSet<>();
    }

    @Override
    public JobParameters getJobParameters(Job job, StepExecution stepExecution) {
        JobParametersBuilder builder = new JobParametersBuilder();
        Map<String, JobParameter> jobParameters = stepExecution.getJobParameters().getParameters();
        ExecutionContext stepExecutionContext = stepExecution.getExecutionContext();
        ExecutionContext jobExecutionContext = stepExecution.getJobExecution().getExecutionContext();

        // copy job parameters from parent job to delegate job
        for (String key : jobParameters.keySet()) {
            builder.addParameter(key, jobParameters.get(key));
        }

        // copy job/step context from parent job/step to delegate job
        for (String key : keys) {
            if (jobExecutionContext.containsKey(key)) {
                builder.addString(key, jobExecutionContext.getString(key));
            } else if (stepExecutionContext.containsKey(key)) {
                builder.addString(key, stepExecutionContext.getString(key));
            } else if (jobParameters.containsKey(key)) {
                builder.addString(key, (String) jobParameters.get(key).getValue());
            }
        }
        return builder.toJobParameters();
    }

    public void setKeys(String[] keys) {
        this.keys = new HashSet<>(Arrays.asList(keys));
    }

}

然后我可以在报告从属步骤中使用上述提取器，如下所示：

<step id="reportingSlaveStep" xmlns="http://www.springframework.org/schema/batch">
        <job ref="reportingSlaveJob" job-parameters-extractor="customJobParametersExtractor"/>
</step>

其中customJobParametersExtractor 是CustomJobParametersExtractor 类型的bean，它传递了我们要复制到reportingSlaveJob 的JobParameters 的所有键。

当我运行上述作业时，有时我会收到一个异常，说 “该作业的作业执行已经在运行”，其他时候我 在 MapExecutionContextDao.java:130 上获得 NullPointerException

发生这种情况的原因是因为没有我的CustomJobParameterExtractor，reportingSlaveJob 启动时是空的JobParameters。 Spring Batch 要创建一个新的作业实例，每次启动reportingSlaveJob 的作业参数都必须不同。使用CustomJobParameterExtractor 也解决了这个问题。