[Batch] Execution 데이터 공유

스프링 배치에서는 데이터를 공유할 수 있습니다.

• BATCH_JOB_EXECUTION
  • Job이 실행되는 동안 시작/종료 시간, job 상태 등을 관리
• BATCH_JOB_EXECUTION_CONTEXT
  • Job이 실행되며 공유해야할 데이터를  직렬 화해 저장
• BATCH_STEP_EXECUTION
  • Step이 실행되는 동안 필요한 데이터 또는 실행된 결과 저장
• BATCH_STEP_EXECUTION_CONTEXT
  • Step이 실행되며 공유해야할 데이터를 직렬 화해 저장

저번 글에서 적은 내용입니다.

위와같이 Job과 Step 에는 EXECUTION과 EXECUTION_CONTEXT 가 있습니다.

두 개가 데이터를 저장하는 역할을 합니다.

여기서 BATCH_JOB_EXECUTION_CONTEXT와 BATCH_STEP_EXECUTION_CONTEXT의 데이터를 공유할 수 있는 차이에 대해서 코드로 알아보겠습니다.

 

@Configuration
@Slf4j
@RequiredArgsConstructor
public class SharedConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    /**
     * start는 시작  그 다음 next 순차적으로 실행된다.
     * @return
     */
    @Bean
    public Job shareJob() {
        return jobBuilderFactory.get("shareJob")
                .incrementer(new RunIdIncrementer())
                .start(this.shareStep())
                .next(this.shareStep2())
                .build();
    }

    /**
     * contribution 을 이용해서 stepExecution 객체를 꺼낼수 있다.
     * stepExecution을 이용해서 stepExecutionContext을 꺼낼수 있다.
     * @return
     */
    @Bean
    public Step shareStep() {
        return stepBuilderFactory.get("shareStep1")
                .tasklet((contribution, chunkContext) -> {
                    // step ExecutionContext.put
                    StepExecution stepExecution = contribution.getStepExecution();
                    ExecutionContext stepExecutionContext = stepExecution.getExecutionContext();
                    stepExecutionContext.putString("step", "step execution context");

                    // job ExecutionContext.put
                    JobExecution jobExecution = stepExecution.getJobExecution();
                    ExecutionContext jobExecutionContext = jobExecution.getExecutionContext();
                    jobExecutionContext.putString("job", "job execution context");

                    // log
                    JobInstance jobInstance = jobExecution.getJobInstance();
                    JobParameters jobParameters = jobExecution.getJobParameters();
//                    JobParameters jobParameters1 = stepExecution.getJobParameters();

                    log.info("jobName : {}, stepName : {}, run.id : {}",
                            jobInstance.getJobName(),
                            stepExecution.getStepName(),
                            jobParameters.getLong("run.id"));

                    return RepeatStatus.FINISHED;
                }).build();
    }

    @Bean
    public Step shareStep2() {
        return stepBuilderFactory.get("shareStep2")
                .tasklet((contribution, chunkContext) -> {
                    // step ExecutionContext.get
                    StepExecution stepExecution = contribution.getStepExecution();
                    ExecutionContext stepExecutionContext = stepExecution.getExecutionContext();

                    // job ExecutionContext.get
                    JobExecution jobExecution = stepExecution.getJobExecution();
                    ExecutionContext jobExecutionContext = jobExecution.getExecutionContext();

                    // log
                    log.info("jobValue : {}, stepValue : {}",
                            jobExecutionContext.getString("job", "emptyJob"),
                            stepExecutionContext.getString("step", "emptyStep"));

                    return RepeatStatus.FINISHED;

                }).build();
    }
}

위와 같은 클래스가 있습니다.

Step에서 tasklet은 다음과 같은 기능을 합니다.

• . tasklet((contribution, chunkContext))
  • Step 안에서 수행될 기능들을 명시합니다.
  • Tasklet은 Step안에서 단일로 수행될 커스텀한 기능들을 선언할때 사용합니다.

shareJob을 보면 start를 통해 shareStep()을 실행 후에 next로 shareStep2()를 순차적으로 실행하는 것을 볼 수 있습니다.

 

그리고 shareStep()에서는 contribution 을 이용해서 stepExecution 객체를 꺼낼 수 있고 -> contribution.getStepExecution();

stepExecution을 이용해서 stepExecutionContext을 꺼낼수 있습니다.-> stepExecution.getExecutionContext();

그리고 꺼낸 stepExecution 에 키 벨류를 넣어줍니다. Job도 같은 방법입니다.

 

stepShare2()에서는 shareStep()에서 만든 ExecutionContext객체에 넣은 값을 가져오도록 만들었습니다.

// log
log.info("jobValue : {}, stepValue : {}",
        jobExecutionContext.getString("job", "emptyJob"),
        stepExecutionContext.getString("step", "emptyStep"));

위의 코드는 키 값으로 job이 있으면 job의 벨류를 가져오고 없으면 emptyJob을 가져오도록 하는 로직입니다.

 

이렇게 한후 Edit Configurations에서 --job.name=shareJob으로 변경해줍니다.

 

실행해보겠습니다. 

 

Executing step [shareStep1] 이 실행된 것을 알 수 있고

jobName 은 jobInstance를 가져왔기때문에 shareJob이 찍혔고

stepName은 shareStep()에서 get한 shareStep1이 찍혔고

run.id는 Job에서 설정한 new RunIdIncrementer로 인해 새로운 아이디로 runId를 가져오는 모습입니다.

 

그리고 Excuting step :[shareStep2] 가 실행된 것을 확인했습니다.

jobValue는 job이란 키를 가진 것을 가져오게 했는데 위의 shareStep에서 넣은 JobExecution을 가져와 job execution context를 가져오고

stepValue는 비어있을 때 나오는 emptyStep이 로그에 찍혔습니다.

 

그러면 shareStep2()에서 값을 넣어 줘 보겠습니다.

 

@Bean
public Step shareStep2() {
    return stepBuilderFactory.get("shareStep2")
            .tasklet((contribution, chunkContext) -> {
                // step ExecutionContext.get
                StepExecution stepExecution = contribution.getStepExecution();
                ExecutionContext stepExecutionContext = stepExecution.getExecutionContext();

                // job ExecutionContext.get
                JobExecution jobExecution = stepExecution.getJobExecution();
                ExecutionContext jobExecutionContext = jobExecution.getExecutionContext();

                StepExecution stepExecution2 = contribution.getStepExecution();
                ExecutionContext stepExecutionContext2 = stepExecution2.getExecutionContext();
                stepExecutionContext2.putString("step", "step execution context");

                // log
                log.info("jobValue : {}, stepValue : {}",
                        jobExecutionContext.getString("job", "emptyJob"),
                        stepExecutionContext.getString("step", "emptyStep"));

                return RepeatStatus.FINISHED;

            }).build();
}

 

empty가 아닌 step execution context가 찍히는 모습입니다.

 이것으로 jobExecution는 job이 관리하는 step 어디에서든 데이터를 공유할 수 있고 stepExecution은 해당 스텝 내에서만 관리 가능하다는 것을 알 수 있습니다.