在运行并行作业时,如何将参数从Tasklet安全地传递到步骤[英] How to safely pass params from Tasklet to step when running parallel jobs

问题描述

我正在尝试将参数从 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**"]}]}]}   

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

推荐答案

查看您尝试的方法:

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

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

代码将更新如下:

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

本文地址:https://www.itbaoku.cn/post/978777.html