我正在尝试将参数从 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**"]}]}]}
知道为什么我会遇到这些线程问题吗?
最佳答案
查看您尝试的方法:
- 方法一——编辑
JobParameters
JobParameters
在作业中是不可变的,因此不应在作业执行期间尝试修改它们。 - 方法二——编辑
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);
关于java - 运行并行作业时如何安全地将参数从 Tasklet 传递到步骤,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/29776901/