Как мы можем делиться данными между различными этапами работы в Spring Batch?

Question

Рассматривая Spring Batch, я хотел бы узнать, как мы можем обмениваться данными между различными этапами работы?

Можем ли мы использовать JobRepository для этого? Если да, то как мы можем это сделать?

Есть ли другой способ сделать / достичь того же?

Answer 1

репозиторий заданий используется косвенно для передачи данных между этапами (Жан-Филипп прав, что лучший способ сделать это - поместить данные в StepExecutionContext, а затем использовать многословное имя ExecutionContextPromotionListener для продвижения ключей контекста выполнения шага к JobExecutionContext.

Полезно отметить, что есть слушатель для продвижения JobParameter ключей к StepExecutionContext (еще более многословно названному JobParameterExecutionContextCopyListener); вы обнаружите, что используете их очень часто, если ваши рабочие шаги не полностью независимы друг от друга.

В противном случае вам остается передавать данные между шагами, используя еще более сложные схемы, такие как очереди JMS или(не дай бог) жестко закодированные местоположения файлов.

Что касается размера данных, которые передаются в контексте, я бы также посоветовал вам сохранять их небольшими (но у меня нет никаких подробностей относительно

Answer 2

На шаге вы можете поместить данные в StepExecutionContext. Затем с помощью прослушивателя вы можете продвигать данные от StepExecutionContext до JobExecutionContext.

Это JobExecutionContext доступно на всех следующих этапах.

Осторожно: данные должны быть короткими. Эти контексты сохраняются в JobRepository сериализацией, а длина ограничена (2500 символов, если я хорошо помню).

Так что эти контексты хороши для совместного использования строк или простых значений, но не для обмена коллекциями или огромными объемами данных.

Обмен огромными объемами данных не является философией Spring Batch. Spring Batch - это набор отдельных действий, а не огромная единица бизнес-обработки.

Answer 3

Я бы сказал, что у вас есть 3 варианта:

1. Используйте StepContext и повышайте его до JobContext, и у вас есть доступ к нему с каждого шага, вы должны, как отмечалось, соблюдать ограничение в размере
2. Создание @JobScope компонента и добавление данных в этот компонент, @Autowire, где это необходимо, и использование его (недостаток заключается в том, что это структура в памяти, и если задание не выполняется, данные теряются, это может вызвать проблемы с перезапуском)
3. У нас были большие наборы данных, которые нужно было обработать по шагам (прочитать каждую строку в csv и записать в БД, прочитать из БД, объединить и отправить в API), поэтому мы решили смоделировать данные в новой таблице в той же БД, что иметатаблицы Spring Batch сохраните ids в JobContext и получите доступ при необходимости и удалите эту временную таблицу после успешного завершения задания.

Answer 4

Вы можете использовать объект Java Bean

1. Выполнить один шаг
2. Сохранить результат в объекте Java
3. Следующий шаг будет ссылаться на тот же Java-объектполучить результат, сохраненный с шагом 1

Таким образом, вы можете хранить огромную коллекцию данных, если хотите

Answer 5

Вот что я сделал, чтобы сохранить объект, доступный по шагам.

1. Создал слушатель для установки объекта в контексте задания

@Component("myJobListener")
public class MyJobListener implements JobExecutionListener {

    public void beforeJob(JobExecution jobExecution) {

        String myValue = someService.getValue();
        jobExecution.getExecutionContext().putString("MY_VALUE", myValue);
    }
}

1. Определяет слушателя в контексте задания

<listeners>
         <listener ref="myJobListener"/>
</listeners>

1. Используется значение в шаге с использованием аннотации BeforeStep

@BeforeStep
public void initializeValues(StepExecution stepExecution) {

String value = stepExecution.getJobExecution().getExecutionContext().getString("MY_VALUE");

}

Answer 6

Вы можете хранить данные в простом объекте. Как:

AnyObject yourObject = new AnyObject();

public Job build(Step step1, Step step2) {
    return jobBuilderFactory.get("jobName")
            .incrementer(new RunIdIncrementer())
            .start(step1)
            .next(step2)
            .build();
}

public Step step1() {
    return stepBuilderFactory.get("step1Name")
            .<Some, Any> chunk(someInteger1)
            .reader(itemReader1())
            .processor(itemProcessor1())
            .writer(itemWriter1(yourObject))
            .build();
}

public Step step2() {
    return stepBuilderFactory.get("step2Name")
            .<Some, Any> chunk(someInteger2)
            .reader(itemReader2())
            .processor(itemProcessor2(yourObject))
            .writer(itemWriter2())
            .build();
}

Просто добавьте данные к объекту в модуле записи или любым другим методом и получите их на любом этапе следующего шага

Answer 7

Использовать `ExecutionContextPromotionListener.

public class YourItemWriter implements ItemWriter<Object> {
private StepExecution stepExecution;
public void write(List<? extends Object> items) throws Exception {
// Some Business Logic

// put your data into stepexecution context
ExecutionContext stepContext = this.stepExecution.getExecutionContext();
stepContext.put("someKey", someObject);
}
@BeforeStep
public void saveStepExecution(Final StepExecution stepExecution) {
this.stepExecution = stepExecution;
}
}

Теперь вам нужно добавить PromotionListener к вашей работе

@Bean
public Step step1() {
        return stepBuilder
        .get("step1")<Company,Company>  chunk(10)
        .reader(reader()).processor(processor()).writer(writer())
        .listener(promotionListener()).build();
    }

@Bean
public ExecutionContextPromotionListener promotionListener() {
    ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
    listener.setKeys(new String[] {"someKey"});
    listener.setStrict(true);
    return listener;
}

Теперь на шаге 2 получите данные из задания. ExecutionContext

public class RetrievingItemWriter implements ItemWriter<Object> {
private Object someObject;
public void write(List<? extends Object> items) throws Exception {
// ...
}
@BeforeStep
public void retrieveInterstepData(StepExecution stepExecution) {
JobExecution jobExecution = stepExecution.getJobExecution();
ExecutionContext jobContext = jobExecution.getExecutionContext();
this.someObject = jobContext.get("someKey");
}
}

Если вы работаете с тасклетами, используйте следующую команду, чтобы получить или поместить ExecutionContext

List<YourObject> yourObjects = (List<YourObject>) chunkContent.getStepContext().getJobExecutionContext().get("someKey");

Answer 8

Мне дали задание вызывать пакетное задание один за другим. Каждое задание зависит от другого. Первый результат работы должен выполнить последующую программу работы. Я искал, как передать данные после выполнения задания. Я обнаружил, что этот ExecutionContextPromotionListener пригодится.

1) Я добавил компонент для «ExecutionContextPromotionListener», как показано ниже

@Bean
public ExecutionContextPromotionListener promotionListener()
{
    ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
    listener.setKeys( new String[] { "entityRef" } );
    return listener;
}

2) Затем я прикрепил одного из слушателей к своим ступеням

Step step = builder.faultTolerant()
            .skipPolicy( policy )
            .listener( writer )
            .listener( promotionListener() )
            .listener( skiplistener )
            .stream( skiplistener )
            .build();

3) Я добавил stepExecution в качестве ссылки в моей реализации шага Writer и заполнил его до шага

@BeforeStep
public void saveStepExecution( StepExecution stepExecution )
{
    this.stepExecution = stepExecution;
}   

4) в конце моего шага писателя я заполнил значения в пошаговом исполнении в виде клавиш, как показано ниже

lStepContext.put( "entityRef", lMap );

5) После выполнения задания я получил значения из lExecution.getExecutionContext() и заполняется как ответ на задание.

6) из объекта ответа задания я получу значения и заполню необходимые значения в остальных заданиях.

Приведенный выше код предназначен для продвижения данных из шагов в ExecutionContext с использованием ExecutionContextPromotionListener. Это можно сделать за любые шаги.

Answer 9

Другой очень простой подход, оставляя здесь для дальнейшего использования:

class MyTasklet implements Tasklet {
    @Override
    public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) {
        getExecutionContext.put("foo", "bar");
    }
}

class MyOtherTasklet implements Tasklet {
    @Override
    public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) {
        getExecutionContext.get("foo");
    }   
}

getExecutionContext здесь:

ExecutionContext getExecutionContext(ChunkContext chunkContext) {
    return chunkContext.getStepContext()
                       .getStepExecution()
                       .getJobExecution()
                       .getExecutionContext();
}     

Поместите его в суперкласс, в интерфейс какdefault метод, или просто вставьте в Tasklet с.

Answer 10

Как сказал Ненад Божич в своем 3-м варианте, использование временных таблиц для совместного использования данных между шагами, использование общего контекста также делает то же самое, он записывает в таблицу и загружает обратно на следующем шаге, но если вы записываете во временные таблицы, выможно убирать в конце работы.