[Spring] Spring Batch 사용해보기(3) - step(tasklet, chunk)

1. Tasklet

- Tasklet 은 하나의 메소드로 구성되어 있는 인터페이스.

- Step 안에서 수행될 기능들을 명시.

- 실패를 알리기위해 예외를 반환하거나 Throw 할 때까지 execute 를 반복적으로 호출.

https://khj93.tistory.com/entry/Spring-Batch%EB%9E%80-%EC%9D%B4%ED%95%B4%ED%95%98%EA%B3%A0-%EC%82%AC%EC%9A%A9%ED%95%98%EA%B8%B0

- Tasklet 은 Step 안에서 단일로 수행도리 커스텀한 기능들을 선언할 때 사용

- Tasklet 은 하나와 Reader & Processor & Writer 한 묶음이 같은 레벨로서, Reader & Processor 와 함께 Tasklet 을 사용할 수 없다.

1-1. Sample(With. Lambda)

@Slf4j
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class TaskletJobConfig{

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job TaskletJob(){

        Job customJob = jobBuilderFactory.get("taskletJob")
                .start(TaskStep())
                .build();

        return customJob;
    }

    @Bean
    public Step TaskStep(){
        return stepBuilderFactory.get("taskletStep")
                .tasklet((contribution, chunkContext) ->{
                    //비즈니스 로직
                    for(int idx = 0; idx < 5; idx ++){
                        log.info("[idx] = " + idx);
                    }
                    return RepeatStatus.FINISHED;
                }).build();
    }

}

1-2. Sample(With. 외부클래스)

@Slf4j
@EnableBatchProcessing
@Configuration
@RequiredArgsConstructor
public class TaskletJobConfig{

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job TaskletJob(){

        Job customJob = jobBuilderFactory.get("taskletJob")
                .start(TaskStep())
                .build();

        return customJob;
    }

    @Bean
    public Step TaskStep(){
        return stepBuilderFactory.get("taskletStep1")
                .tasklet(new CustomTasklet())
                .build();
    }

}

@Slf4j
public class CustomTasklet implements Tasklet, StepExecutionListener {

    @Override
    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        log.info("Before Step Start!");
    }

    @Override
    @AfterStep
    public ExitStatus afterStep(StepExecution stepExecution) {

        log.info("After Step Start!");

        return ExitStatus.COMPLETED;
    }

    @Override
    public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception {
        //비즈니스 로직
        for (int idx = 0; idx < 5; idx++) {
            log.info("[idx] = " + idx);
        }
        return RepeatStatus.FINISHED;
    }
}

2. Chunk

- 여러 개의 아이템을 묶은 하나의 덩어리를 의미.

- 처리되는 Commit Row 의 수

- 한 번에 하나의 아이템을 입력받아 Chunk 단위로 덩어리를 만들어 Chunk 단위로 Transaction(트랜잭션)을 처리.

Chunk 단위로 Commit 과 Rollback 이 동작

- 대용량 데이터를 한 번에 처리하는 것이 아닌 Chunk 단위로 쪼개어 반복 수행.

https://khj93.tistory.com/entry/Spring-Batch%EB%9E%80-%EC%9D%B4%ED%95%B4%ED%95%98%EA%B3%A0-%EC%82%AC%EC%9A%A9%ED%95%98%EA%B8%B0

2-1. Chunk <I>, Chunk <O>

• Chunk<I>
  - ItemRead 로 읽은 하나의 아이템을 Chunk 크기만큼 반복해서 저장
• Chunk<O>
  - ItemReader 로 부터 전달받은 Chunk<I> 를 참조해서 ItemProcessor 에서 가공하여 ItemWriter 로 전달

2-2. 아키텍쳐

https://backtony.github.io/spring/2022-01-24-spring-batch-6/

• ItemReader 가 데이터를 한 건씩 읽고 한 건씩 Chunk 크기만큼 Chunk<I> 에 저장.
• Chunk 크기 만큼 쌓였다면 Chunk<I> 를 ItemProcessor 에 전달.
• ItemProcessor 는 전달받은 Chunk 를 적절하게 가공해서 Chunk<O> 에 저장.
• Chunk<O> 를 ItemWriter 에 전달.
• ItemWriter 는 데이터를 쓰기 작업.

ItemReader 와 ItemProcessor 는 각각의 하나씩 아이템을 처리하지만, ItemWriter 는 Chunk 크기만큼을 한 번에 일괄 처리한다.

2-3. sample(with. JPA)

@Slf4j
@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
public class ChunkJdbcJobConfig {
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final EntityManagerFactory entityManagerFactory;

    @Bean
    public Job ChunkJob() {
        Job chunkJob = jobBuilderFactory.get("ChunkJob0")
                .start(step())
                .build();

        return chunkJob;
    }

    @Bean
    @JobScope
    public Step step() {
        return stepBuilderFactory.get("step")
                .<Member, Member>chunk(5)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }

    @Bean
    @StepScope
    public JpaPagingItemReader<Member> reader() {

        Map<String,Object> parameterValues = new HashMap<>();
        log.info("ItemReader 실행됌");
        return new JpaPagingItemReaderBuilder<Member>()
                .pageSize(5)
                .parameterValues(parameterValues)
                .queryString("SELECT m FROM Member m")
                .entityManagerFactory(entityManagerFactory)
                .name("JpaPagingItemReader")
                .build();
    }

    @Bean
    @StepScope
    public ItemProcessor<Member, Member> processor(){
        log.info("ItemProcessor 실행됌");
        return new ItemProcessor<Member, Member>() {
            @Override
            public Member process(Member member) throws Exception {
                log.error(" ID : " + member.getId() + " / Name : " + member.getName() + " / amount : " + member.getAmount());
                member.setAmount(member.getAmount() + 1000);
                log.info(" ID : " + member.getId() + " / Name : " + member.getName() + " / amount : " + member.getAmount());
                return member;
            }
        };
    }

    @Bean
    @StepScope
    public JpaItemWriter<Member> writer(){
        log.info("ItemWriter 실행됌");
        return new JpaItemWriterBuilder<Member>()
                .entityManagerFactory(entityManagerFactory)
                .build();
    }
}

---

참고

- https://khj93.tistory.com/entry/Spring-Batch%EB%9E%80-%EC%9D%B4%ED%95%B4%ED%95%98%EA%B3%A0-%EC%82%AC%EC%9A%A9%ED%95%98%EA%B8%B0