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

我试图安全地将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 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 – 编辑JobParameters v2方法2与方法1非常相似，您只是以不同的方式获取对JobParameters的引用。
• 方法3 – 使用ExecutionContextPromotionListener 。 这是正确的方法，但你做错了。 ExecutionContextPromotionListener查看步骤的 ExecutionContext并将指定的键复制到作业的ExecutionContext 。 您将密钥直接添加到Job ExecutionContext，这是一个坏主意。

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

代码将更新如下：

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