스프링 배치(Spring Batch) 활용하기

앞서 Spring Batch를 소개하고 간단한 예제를 작성하고 실행해보 Meta Data엔 어떤 정보들이 있는지 알아보았다. 이번에는 실전에서 사용할 수 있는 간단한 Spring Batch 내용들을 실행해보며 정리하겠다.

 

Next

next()는 step1 -> step2 -> stpe3 순으로 하나씩 Step을 실행시킬때 사용한다.

@Slf4j
@RequiredArgsConstructor
@Configuration
public class NextSimpleJobConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job nextSimpleJob() {
        return jobBuilderFactory.get("nextSimpleJob")
            .start(nextSimpleStep1())
            .next(nextSimpleStep2())
            .build();
    }

    @Bean
    public Step nextSimpleStep1() {
        return stepBuilderFactory.get("nextSimpleStep1")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is nextSimpleStep1");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Step nextSimpleStep2() {
        return stepBuilderFactory.get("nextSimpleStep2")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is nextSimpleStep2");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

}

---

지정한 Batch Job만 실행되도록 설정

Batch Job을 여러개 생성하고 특정한 Job만 실행되도록 하고싶다면 application.yml 파일에 아래 코드를 추가한다.

spring.batch.job.names: ${job.name:NONE}

 

이 코드의 의미는 Program arguments에 job.name이 있으면 job.name값을 할당하고, 없으면 NONE을 할당하겠다는 의미다. 여기서 NONE이 할당 된다면 아무 배치도 실행되지 않는다.

---

조건 별 흐름 제어 (Flow)

Step이 진행될때 상황에 따라 조건을 걸어 다른 Step을 실행할 때가 있다.

예를 들어, Step A, B, C가 실행될때 정상일때는 Step B로, 오류가 났을때는 Step C로 수행해야 한다.

이럴 경우를 대비해 Spring Batch Job에서는 조건별로 Step을 사용할 수 있다.

@Slf4j
@Configuration
@RequiredArgsConstructor
public class StepNextConditionalJobConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job stepNextConditionalJob() {
        return jobBuilderFactory.get("stepNextConditionalJob")
            .start(stepNextConditionalStepA()) // StepA 실행
                .on("FAILED") // ExitStatus가 FAILED 일 경우
                .to(stepNextConditionalStepC()) // StepC 실행
                .on("*") // 결과 관계 없이
                .end() // Flow 종료
            .from(stepNextConditionalStepA()) // StepA로 부터
                .on("*") // FAILED 외에 모든 경우
                .to(stepNextConditionalStepB()) // StepB 실행
                .next(stepNextConditionalStepC()) // StepB가 정상 종료되면 StepC 실행
                .on("*") // 결과 관계 없이
                .end() // Flow 종료
            .end() // Job 종료
            .build();
    }

    @Bean
    public Step stepNextConditionalStepA() {
        return stepBuilderFactory.get("stepNextConditionalStepA")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is stepNextConditionalStepA");
                // ExitStatus에 따라 Flow가 진행된다.
                contribution.setExitStatus(ExitStatus.FAILED);
                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Step stepNextConditionalStepB() {
        return stepBuilderFactory.get("stepNextConditionalStepB")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is stepNextConditionalStepB");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Step stepNextConditionalStepC() {
        return stepBuilderFactory.get("stepNextConditionalStepC")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is stepNextConditionalStepC");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

}

위 코드의 시나리오는 아래와 같다.

• StepA 실행 실패 -> StepC 실행 -> Flow 종료
  

  

• StepA 실행 성공 -> StepB 실행 성공 -> StepC 실행 -> Flow 종료
  

  

 

BatchStatus와 ExitStatus

위에서 Step에 결과상태는 BatchStatus가 아닌 ExitStatus로 판단한다.

BatchStatus와 ExitStatus의 차이점을 알고 있어야 한다.

• BatchStatus : Job 또는 Step 의 실행 결과를 Spring에서 기록할 때 사용
• ExitStatus :  Step의 실행 후 상태

Spring Batch는 기본적으로 ExitStatus의 exitCode는 Step의 BatchStatus와 같도록 설정이 되어 있다.

하지만 만약에 본인만의 커스텀한 exitCode가 필요하다면 별도의 로직이 필요하다.

public class CustomStepExecutionListener extends StepExecutionListenerSupport {

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        ExitStatus exitStatus = stepExecution.getExitStatus();
        if (!exitStatus.equals(ExitStatus.FAILED)) {
            return new ExitStatus("CUSTOM EXIT CODE");
        } else {
            return null;
        }
    }

}

위 코드에 의미는 Step이 실행된 후 ExitStatus가 FAILED가 아닐때 CUSTOM EXIT CODE를 반환한다.

그럼 이제 StepD를 추가하고, StepA의 결과가 CUSTOM EXIT CODE일때 StepD가 실행되도록 설정해보겠다.

    // ... 생략 
    
    @Bean
    public Job stepNextConditionalJob() {
        return jobBuilderFactory.get("stepNextConditionalJob")
            .start(stepNextConditionalStepA()) // StepA 실행
                .on("FAILED") // 실행 결과가 FAILED 일 경우
                .to(stepNextConditionalStepC()) // StepC 실행
                .on("*") // 결과 관계 없이
                .end() // Flow 종료
            .from(stepNextConditionalStepA()) // StepA로 부터
                .on("CUSTOM EXIT CODE") // 실행 결과가 CUSTOM EXIT CODE 일 경우
                .to(stepNextConditionalStepD()) // SteD 실행
                .on("*") // 결과 관계 없이
                .end() // Flow 종료
            .from(stepNextConditionalStepA()) // StepA로 부터
                .on("*") // 실행 결과가 FAILED, CUSTOM EXIT CODE 외에 모든 경우
                .to(stepNextConditionalStepB()) // StepB 실행
                .next(stepNextConditionalStepC()) // StepB가 정상 종료되면 StepC 실행
                .on("*") // 결과 관계 없이
                .end() // Flow 종료
            .end() // Job 종료
            .build();
    }

    @Bean
    public Step stepNextConditionalStepA() {
        return stepBuilderFactory.get("stepNextConditionalStepA")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is stepNextConditionalStepA");
                return RepeatStatus.FINISHED;
            })
            .listener(new CustomStepExecutionListener()) // listener 설정
            .build();
    }

	// ... 생략 

    @Bean
    public Step stepNextConditionalStepD() {
        return stepBuilderFactory.get("stepNextConditionalStepD")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is stepNextConditionalStepD");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

	// ... 생략

실행 결과 StepA 실행 후 StepD가 실행되었다.

---

Decide

위에서 진행했던 조건 별 흐름 제어에는 2가지 문제가 있다.

 

Step이 담당하는 역할이 2개 이상이 된다.

실제 해당 Step이 처리해야할 로직외에도 분기처리를 시키기 위해 ExitStatus 조작이 필요

다양한 분기 로직 처리의 어려움

ExitStatus를 커스텀하게 고치기 위해선 Listener를 생성하고 Job Flow에 등록하는 등 번거로움

 

그래서 Spring Batch에서는 Step들의 Flow속에서 분기만 담당하는 JobExecutionDecider이 있다.

@Slf4j
@Configuration
@RequiredArgsConstructor
public class DeciderJobConfiguration {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job deciderJob() {
        return jobBuilderFactory.get("deciderJob")
            .start(startStep()) // startStep 실행
            .next(decider()) // Step 분기
            .from(decider()) // decider 결과가
                .on("StepA") // StepA 라면
                .to(stepA()) // StepA 실행
            .from(decider()) // decider 결과가
                .on("StepB") // StepB 라면
                .to(stepB()) // StepB 실행
            .end()
            .build();
    }

    @Bean
    public Step startStep() {
        return stepBuilderFactory.get("StartStep")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> Start!");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Step stepA() {
        return stepBuilderFactory.get("StepA")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is StepA");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public Step stepB() {
        return stepBuilderFactory.get("StepB")
            .tasklet((contribution, chunkContext) -> {
                log.info(">>>>> This is StepB");
                return RepeatStatus.FINISHED;
            })
            .build();
    }

    @Bean
    public JobExecutionDecider decider() {
        return new StepDecider();
    }

    public static class StepDecider implements JobExecutionDecider {
        @Override
        public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) {
            String stepName = "StepA";
            return new FlowExecutionStatus(stepName);
        }
    }

}

코드를 보면 분기 로직에 대한 모든 일은 StepDecider가 전담하고 있다.

아무리 복잡한 분기로직이 필요하더라도 Step과는 명확히 역할과 책임이 분리된채 진행할 수 있다.

실행 결과 분기가 정삭적으로 작동하는 것을 확인할 수 있다.

 

---

※ 전체 코드는 GitHub에서 확인하실 수 있습니다.

※ 참고 블로그 https://jojoldu.tistory.com/328