Как безопасно передавать параметры от Tasklet до шага при выполнении параллельных заданий

Я пытаюсь передать безопасные параметры из tasklet на шаг в той же работе.

Моя работа состоит из 3-х цепей (step1, step2, step3) один за другим и в конце шаг 4 (процессор, считыватель, запись)

эта работа выполняется много раз параллельно.

В шаге 1 внутри tasklet я оцениваю param (hashId) через веб-службу), чем передаю его по всей цепочке до моего читателя (что на шаге 4)

На шаге 3 я создаю новый параметр: filePath, основанный на hashid, и я отправляю его на step4 (читатель) в качестве местоположения файлового ресурса

Я использую stepExecution для передачи этого параметра (hashId и filePath).

Я попробовал 3 способа сделать это через тасклет:

передать параметр (hashId с шага 1 на шаг 2 и с шага 2 на шаг 3) Я делаю это:

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

В шаге 4 я заполняю filePath на основе hashId и передаю его таким образом на мой последний шаг (который является процессором чтения и писателем).

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 раз, я вижу, что значение параметра paramPath заполняется другими значениями filePath заданий при параллельном выполнении

Вот как я продвигаю ключи задания с помощью executeContextPromotionListener:

определение задания:

 @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;
    }

Те же результаты, что и во введении params

Я могу отслеживать результаты через spring таблицу базы данных пакетов: batch_job_execution_context.short_context:

здесь вы можете увидеть, что filePatch, созданный хешидом, не идентичен исходному hashId // неправильная запись ///

{ "карта": [{ "входа": [{ "строка" : "totalRecords", "Int": 5}, { "строка" : "segmentId", "длинный": 13}, { "строка": [ "Filepath" "/и т.д./MYDIR/услуги/notification_processor/файлы/2015_04_22/ f1c7b0f2180b7e266d36f87fcf6fb7aa.csv" ]}, { "строка" : [ "Хашид", " 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**"]}]}]}

Любая идея, почему у меня есть эти проблемы Threading?

Ответ 1