23. Repeat

  Spring Batch는 특정 조건이 충족될 때까지 (또는 특정 조건이 아직 충족되지 않을 때까지) Job 또는 Step을 반복하도록 배치 애플리케이션을 구성할 수 있습니다. Spring Batch에서는 Step의 반복과 Chunk 반복을 RepeatOperation을 사용해서 처리하며, 기본 구현체로 RepeatTemplate를 제공합니다.

 

반복을 종료할 것인지 여부를 결정하는 세가지 항목

RepeatStatus

• 스프링 배치의 처리가 끝났는지 판별하기 위한 열거형(enum)
• CONTINUABLE: 작업이 남아 있음, FINISHED: 더 이상의 반복 없음

 

CompletionPolicy

• RepeatTemplate의 iterate 메소드 안에서 반복을 중단할지 결정
• 실행 횟수 또는 완료시기, 오류 발생 시 수행할 작업에 대한 반복여부 결정
• 정상 종료를 알리는데 사용

구현 클래스

• SimpleCompletionPolicy: 현재 반복 횟수가 Chunk 개수보다 크면 반복 종료
• CountingCompletionPolicy: 일정한 카운트를 계산 및 집계해서 카운트 제한 조건이 만족하면 반복 종료
• SimpleCompletionPolicy: 반복시점부터 현재시점까지 소요된 시간이 설정된 시간보다 크면 반복 종료

 

ExceptionHandler

• RepeatCallback 안에서 예외가 발생하면 RepeatTemplate가 ExceptionHandler를 참조해서 예외를 다시 던질지 여부 결정
• 예외를 받아서 다시 던지게 되면 반복 종료
• 비정상 종료를 알리는데 사용

구현 클래스

• SimpleLimitExceptionHandler: 예외 타입 중 하나가 발견되면 카운터가 증가하고 한계가 초과되었는지 여부를 확인하고 Throwable을 다시 던짐
• LogOrRethrowExceptionHandler: 예외를 로그로 기록할지 아니면 다시 던질 지 결정
• RethrowOnThresholdExceptionHandler: 지정된 유형의 예외가 임계 값에 도달하면 다시 발생

package io.springbatch.springbatchlecture;

import lombok.RequiredArgsConstructor;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.*;
import org.springframework.batch.item.adapter.ItemWriterAdapter;
import org.springframework.batch.item.support.ClassifierCompositeItemProcessor;
import org.springframework.batch.item.support.CompositeItemProcessor;
import org.springframework.batch.item.support.builder.CompositeItemProcessorBuilder;
import org.springframework.batch.repeat.CompletionPolicy;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler;
import org.springframework.batch.repeat.policy.CompositeCompletionPolicy;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.policy.TimeoutTerminationPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.classify.PatternMatchingClassifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@RequiredArgsConstructor
@Configuration
public class RepeatConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job job() throws Exception {
        return jobBuilderFactory.get("batchJob")
                .incrementer(new RunIdIncrementer())
                .start(step1())
                .build();
    }

    @Bean
    public Step step1() throws Exception {
        return stepBuilderFactory.get("step1")
                .<String, String>chunk(5)
                .reader(new ItemReader<String>() {
                    int i = 0;
                    @Override
                    public String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                        i++;
                        return i > 3 ? null : "item" + i;
                    }
                })
                .processor(new ItemProcessor<String, String>() {

                    RepeatTemplate template = new RepeatTemplate();

                    @Override
                    public String process(String item) throws Exception {

                        // 반복할 때마다 count 변수의 값을 1씩 증가
                        // count 값이 chunkSize 값보다 크거나 같을 때 반복문 종료
//                        template.setCompletionPolicy(new SimpleCompletionPolicy(2));
                        // 소요된 시간이 설정된 시간보다 클 경우 반복문 종료
//                        template.setCompletionPolicy(new TimeoutTerminationPolicy(3000));

                        // 여러 유형의 CompletionPolicy 를 복합적으로 처리함
                        // 여러 개 중에 먼저 조건이 부합하는 CompletionPolicy 에 따라 반복문이 종료됨
//                        CompositeCompletionPolicy completionPolicy = new CompositeCompletionPolicy();
//                        CompletionPolicy[] completionPolicies = new CompletionPolicy[]{new TimeoutTerminationPolicy(3000),new SimpleCompletionPolicy(2)};
//                        completionPolicy.setPolicies(completionPolicies);
//                        template.setCompletionPolicy(completionPolicy);

                        // 예외 제한 횟수만큼 반복문 실행
//                        template.setExceptionHandler(simpleLimitExceptionHandler());

                        template.iterate(new RepeatCallback() {

                            public RepeatStatus doInIteration(RepeatContext context) {
                               System.out.println("repeatTest");
//                               throw new RuntimeException("Exception is occurred");
                                return RepeatStatus.CONTINUABLE;
                            }

                        });

                        return item;
                    }
                })
                .writer(new ItemWriter<String>() {
                    @Override
                    public void write(List<? extends String> items) throws Exception {
                        System.out.println(items);
                    }
                })
                .build();
    }

    @Bean
    public SimpleLimitExceptionHandler simpleLimitExceptionHandler(){
        return new SimpleLimitExceptionHandler(2);
    }
}

 

[참고자료]

인프런-스프링 배치 - Spring Boot 기반으로 개발하는 Spring Batch