java - 运行并行作业时如何安全地将参数从 Tasklet 传递到步骤

Question

我正在尝试将参数从 tasklet 安全地传递到同一作业中的步骤。

我的工作包括 3 个 tasklet(step1,step2,step3) 一个接一个，最后一个 step4(processor,reader,writer)

这个作业被并行执行了很多次。

在 tasklet 内部的第 1 步中，我正在通过 Web 服务评估 param(hashId))，而不是通过我的链传递它直到我的读者(在第 4 步)

在第 3 步中，我创建了一个名为:filePath 的新参数，它基于 hashid，并将其作为文件资源位置发送到第 4 步(阅读器)

我正在使用 stepExecution 来传递这个参数(hashId 和 filePath)。

我通过 tasklet 尝试了 3 种方法:

将 param(hashId 从 step1 传递到 step2 并从 step2 传递到 step 3) 我这样做:

chunkContext.getStepContext()
        .getStepExecution()
        .getExecutionContext()
        .put("hashId", hashId);

在第 4 步中，我根据 hashId 填充 filePath 并将其传递给我的最后一步(即读取器处理器和写入器)

public class DownloadFileTasklet implements Tasklet, StepExecutionListener {
..

    @Override
     public RepeatStatus execute(ChunkContext chunkContext, ExecutionContext    
     executionContext) throws IOException {

    String hashId = chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext().get("hashId");

          ...

filepath="...hashId.csv";
//I used here executionContextPromotionListener in order to promote those keys

        chunkContext.getStepContext()
        .getStepExecution()
        .getExecutionContext()
        .put("filePath", filePath);
    } 

logger.info("filePath + "for hashId=" + hashId);

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

请注意，在完成该步骤(步骤 3)之前，我正在打印 hashId 和 filePath 值。根据日志，它们是一致的并按预期填充

我还在阅读器中添加了日志，以查看我获得的参数。

@Bean
    @StepScope
    public ItemStreamReader<MyDTO> reader(@Value("#{jobExecutionContext[filePath]}") String filePath) {
              logger.info("test filePath="+filePath+");

        return itemReader;
    }

当我执行这个作业约 10 次时，我可以看到 param filePath 值在并行执行时填充了其他作业的 filePath 值

这就是我使用 executionContextPromotionListener 提升作业键的方式:

工作定义:

 @Bean
    public Job processFileJob() throws Exception {
        return this.jobs.get("processFileJob").
                start.(step1).
                next(step2)
                next(downloadFileTaskletStep()). //step3
                next(processSnidFileStep()).build();  //step4

    }

第 3 步定义

  public Step downloadFileTaskletStep() {
        return this.steps.get("downloadFileTaskletStep").tasklet(downloadFileTasklet()).listener(executionContextPromotionListener()).build();
    }


  @Bean
    public org.springframework.batch.core.listener.ExecutionContextPromotionListener executionContextPromotionListener() {
        ExecutionContextPromotionListener executionContextPromotionListener = new ExecutionContextPromotionListener();
        executionContextPromotionListener.setKeys(new String[]{"filePath"});
        return executionContextPromotionListener;
    }

相同的结果线程弄乱了参数

我可以通过spring批处理数据库表跟踪结果:batch_job_execution_context.short_context:

在这里你可以看到由 hashid 构建的 filePatch 与 origin hashId 不同//错误记录///

{"map":[{"entry":[{"string":"totalRecords","int":5},{"string":"segmentId","long":13},{"string ":["filePath","/etc/mydir/services/notification_processor/files/2015_04_22/f1c7b0f2180b7e266d36f87fcf6fb7aa.csv"]},{"string":["hashId","20df39d201fffc7444423cfdf2f43789 "]}]}]}

现在，如果我们检查其他记录，它们看起来不错。但总是一两个搞砸了

//正确的记录

{"map":[{"entry":[{"string":"totalRecords","int":5},{"string":"segmentId","long":13},{"string":["filePath","\/etc\/mydir\/services\/notification_processor\/files\/2015_04_22\/**c490c8282628b894727fc2a4d6fc0cb5**.csv"]},{"string":["hashId","**c490c8282628b894727fc2a4d6fc0cb5**"]}]}]}

{"map":[{"entry":[{"string":"totalRecords","int":5},{"string":"segmentId","long":13},{"string":["filePath","\/etc\/mydir\/services\/notification_processor\/files\/2015_04_22\/**2b21d3047208729192b87e90e4a868e4**.csv"]},{"string":["hashId","**2b21d3047208729192b87e90e4a868e4**"]}]}]}   

知道为什么我会遇到这些线程问题吗？

Answer 1

查看您尝试的方法:

• 方法一——编辑JobParametersJobParameters 在作业中是不可变的，因此不应在作业执行期间尝试修改它们。
• 方法二——编辑JobParameters v2方法 2 与方法 1 完全相同，只是以不同的方式获取对 JobParameters 的引用。
• 方法 3 - 使用 ExecutionContextPromotionListener。这是正确的方法，但你做错了。 ExecutionContextPromotionListener 查看步骤的 ExecutionContext 并将您指定的键复制到作业的ExecutionContext。您将 key 直接添加到 Job ExecutionContext 这是一个坏主意。

简而言之，方法 3 最接近正确，但您应该将要共享的属性添加到步骤的 ExecutionContext 中，然后将 ExecutionContextPromotionListener 配置为将相应的键提升到 Job 的 ExecutionContext。

代码将更新如下:

chunkContext.getStepContext()
            .getStepExecution()
            .getExecutionContext()
            .put("filePath", filePath);