如何在运行并行作业时安全地将params从Tasklet传递到步骤

Question

我试图安全地将params从tasklet传递到同一个工作中的一个步骤.

我的工作包括3个tasklet(第1步,第2步,第3步),最后是step4(处理器,读者,作者)

这项工作正在多次并行执行.

在tasklet里面的第一步我通过web服务评估param(hashId),而不是把它传遍整个链,直到我的读者(在第4步)

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

我正在使用stepExecution传递此参数(hashId和filePath).

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

传递param(hash1d从step1进入step2,从step2进入步骤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;
}

请注意我在完成该步骤之前打印hashId和filePath值(步骤3).通过日志,它们是按照预期一致和填充的

我还在我的读者中添加了日志,以查看我得到的参数.

@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与原始hashId //错误记录///不同

{ "地图":[{ "条目":[{ "串": "总记录", "INT":5},{ "串": "segmentId", "长":13},{ "串":[ "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

要查看您尝试过的方法：

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

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

代码将更新如下：

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