Tasklet, Chunk 비교

Tasklet, Chunk 비교

• Tasklet을 사용한 Task 기반 처리
  
  • Chunk보다 코딩하기 쉽다.
  • 배치 처리 과정이 비교적 쉬운 경우 사용한다.
  • 대량 처리를 하는 경우에는 더 복잡하다.
  • 하나의 큰 덩어리를 여러 덩어리로 나누어 처리하기에는 부적합하다.
• Chunk를 사용한 chunk(덩어리) 기반 처리
  • ItemReader, ItemProcessor, ItemWriter의 관계를 이해할 필요가 있다.
  • 대량 처리를 하는 경우 Tasklet 보다 비교적 쉽게 구현할 수 있다.
  • Ex> 10000개의 데이터를 1000개씩 10개의 덩어리로 나누어 실행

@JobScope, @StepScope

• @Scope는 어떤 시점에 bean을 생성/소멸 시킬지 bean의 lifecycle을 설정할 수 있다.
• @JobScope는 Job이 실행/소멸 시점에 따라 실행/소멸된다. --> step bean에 설정 가능하다.
• @StepScope는 Step이 실행/소멸 시점에 따라 실행/소멸된다. --> tasklet, ItemReader, ItemProcessor, ItemWriter에 설정할 수있다.
• 그럼 왜 스코프를 설정해야 할까?
  • 여러 스탭에서 하나의 tasklet을 동시에 실행한다면? -> 하나의 tasklet은 thread safe 하지않다.
  • 하지만 tasklet에 @StepScope를 설정해 둔다면? -> 스탭에서 tasklet을 실행할 때마다 새로운 lifecycle을 가진 새로운 tasklet을 실행하게 되기 때문에 thread safe하게 처리할 수 있다.
• @Value("#{jobParameters[key]}")를 사용하기 위해서는 @JobScope와 @StepScope는 필수이다.
  • jobParameters에 대한 접근이 스코프에 의해서 접근하기 때문이다.
    
    
    

chunk 기반 task처리

@Bean
public Job chunkProcessingJob() {
    return jobBuilderFactory.get("chunkProcessingJob")
            .incrementer(new RunIdIncrementer())
            .start(this.taskBaseStep())
            .next(this.chunkBaseStep(null))
            .build();
}

@Bean
@JobScope // job 실행 시점에 생성/소멸 ( Step 에 선언 )
public Step chunkBaseStep(@Value("#{jobParameters[chunkSize]}") String chunkSize) {
    return stepBuilderFactory.get("chunkBaseStep")
            .<String, String>chunk(StringUtils.isNotEmpty(chunkSize) ? Integer.parseInt(chunkSize) : 10)
            .reader(itemReader())
            .processor(itemProcessor())
            .writer(itemWriter())
            .build();
}

private ItemWriter<String> itemWriter() {
    return items -> log.info("chunk item size : {}", items.size());
}

private ItemProcessor<String, String> itemProcessor() {
    return item -> item + ", String Batch";
}

private ItemReader<String> itemReader() {
    return new ListItemReader<>(getItems());
}

private List<String> getItems() {
    List<String> items = new ArrayList<>();

    for(int i = 0; i < 100; i++) {
        items.add(i + " Hello");
    }

    return items;
}

• chunkSize 를 10으로 설정 -> 10개씩 나눈다는 의미
• <String, String>은 <inputType, outputType>을 뜻하는데 inputType은 ItemReader로 부터 받은 데이터를 말한다.
• 만약 processor에서 null이 반환되면 writer로 넘어가지 않는다.
• chunk 기반 실행이 종료되는 시점은 ItemReader에서 null을 반환할 때까지 이다.

@Bean
@StepScope
public Tasklet tasklet(@Value("#{jobParameters[chunkSize]}") String value) {
    List<String> items = getItems();
    return (contribution, chunkContext) -> {
        StepExecution stepExecution = contribution.getStepExecution();

        int chunkSize = StringUtils.isNotEmpty(value) ? Integer.parseInt(value) : 10;
        int fromIndex = stepExecution.getReadCount();
        int toIndex = fromIndex + chunkSize;

        if (fromIndex >= items.size()) {
            return RepeatStatus.FINISHED;
        }

        List<String> subList = items.subList(fromIndex, toIndex);

        log.info("task item size : {}", subList.size());

        stepExecution.setReadCount(toIndex);

        return RepeatStatus.CONTINUABLE;
    };
}

• 위와 같이 tasklet기반으로도 chunk 기반 처럼 데이터를 처리할 수 있다. -> 하지만 선호하지 않는다.
• RepeatStatus.FINISHED -> 해당 task를 끝내라는 뜻이다.
• epeatStatus.CONTINUABLE -> task를 반복하라는 뜻이다.
• 이렇게 하면 chunk로 10개씩 나눠서 실행한 거랑 같은 결과를 만들 수 있다.
• Parameter를 JobParameters 또는 Spring EL 방식으로 처리할 수 있다.
  • String parameter = jobParameters.getString(key, defaultValue);

    //배치 파라미터를 이용한 chunkSize 설정
    JobParameters jobParameters = stepExecution.getJobParameters();
    String value = jobParameters.getString("chunkSize", "10");

  
  
  • @Value("#{jobParameters[key]}")

    @Value("#{jobParameters[chunkSize]}")

• 위와 같이 chunkSize를 지정해준다.