[Spring Batch] 5. 실행 제어와 흐름

지금까지는 Step이 순서대로 실행되는 단순한 흐름만 봤다. 실무에서는 "이 Step이 실패하면 다른 Step을 실행하라", "조건에 따라 분기하라" 같은 복잡한 흐름 제어가 필요하다. Spring Batch의 Flow API가 이 역할을 담당한다.

순차 실행

가장 기본적인 흐름이다. Step이 순서대로 실행되고, 하나라도 실패하면 Job 전체가 실패한다.

@Bean
public Job simpleJob(JobRepository jobRepository,
                      Step stepA, Step stepB, Step stepC) {
    return new JobBuilder("simpleJob", jobRepository)
            .start(stepA)
            .next(stepB)
            .next(stepC)
            .build();
}

조건부 분기

Step의 결과에 따라 다음에 실행할 Step을 결정한다. ExitStatus 값을 기준으로 분기한다.

@Bean
public Job conditionalJob(JobRepository jobRepository,
                            Step validateStep, Step processStep,
                            Step errorStep, Step notifyStep) {
    return new JobBuilder("conditionalJob", jobRepository)
            .start(validateStep)
                .on("FAILED").to(errorStep)     // 검증 실패 → 에러 처리
                .from(validateStep)
                .on("*").to(processStep)         // 그 외 → 정상 처리
                .from(processStep)
                .on("*").to(notifyStep)          // 처리 완료 → 알림
            .end()
            .build();
}

on("FAILED")은 ExitStatus가 "FAILED"일 때 매칭된다. on("*")은 와일드카드로 모든 상태에 매칭된다. 매칭 순서가 중요한데, 먼저 선언된 조건이 우선한다.

ExitStatus vs BatchStatus

혼동하기 쉬운 두 가지 상태가 있다.

• BatchStatus : Spring Batch가 관리하는 실행 상태다. COMPLETED, FAILED, STOPPED 등 고정된 값이다.
• ExitStatus : Step이 반환하는 종료 코드다. 개발자가 커스텀 값을 만들 수 있다.

@Bean
public Step validateStep(JobRepository jobRepository,
                          PlatformTransactionManager transactionManager) {
    return new StepBuilder("validateStep", jobRepository)
            .tasklet((contribution, chunkContext) -> {
                boolean isValid = validateData();
                if (!isValid) {
                    contribution.setExitStatus(new ExitStatus("INVALID"));
                }
                return RepeatStatus.FINISHED;
            }, transactionManager)
            .build();
}

ExitStatus("INVALID")처럼 커스텀 상태를 만들어서 분기에 활용할 수 있다.

ExecutionContext — Step 간 데이터 공유

Step 사이에서 데이터를 주고받아야 할 때 ExecutionContext를 사용한다. 키-값 저장소처럼 동작하며, 메타데이터 테이블에 영속화되므로 재시작 후에도 값이 유지된다.

// Step 1에서 데이터 저장
@Bean
public Step countStep(JobRepository jobRepository,
                       PlatformTransactionManager transactionManager) {
    return new StepBuilder("countStep", jobRepository)
            .tasklet((contribution, chunkContext) -> {
                int count = orderRepository.countPending();
                chunkContext.getStepContext()
                    .getStepExecution()
                    .getJobExecution()
                    .getExecutionContext()
                    .putInt("totalCount", count);
                return RepeatStatus.FINISHED;
            }, transactionManager)
            .build();
}

// Step 2에서 데이터 읽기
@Bean
@StepScope
public ItemReader<Order> orderReader(
        @Value("#{jobExecutionContext['totalCount']}") int totalCount) {
    log.info("처리할 총 건수: {}", totalCount);
    // ...
}

jobExecutionContext는 Job 전체에서 공유되는 컨텍스트다. stepExecutionContext는 해당 Step 내에서만 유효하다.

Flow — 재사용 가능한 Step 묶음

여러 Job에서 동일한 Step 시퀀스를 재사용하고 싶을 때 Flow로 묶는다.

@Bean
public Flow validationFlow(Step schemaCheck, Step dataCheck) {
    return new FlowBuilder<SimpleFlow>("validationFlow")
            .start(schemaCheck)
            .next(dataCheck)
            .build();
}

@Bean
public Job job1(JobRepository jobRepository,
                 Flow validationFlow, Step processStep) {
    return new JobBuilder("job1", jobRepository)
            .start(validationFlow)
            .next(processStep)
            .end()
            .build();
}

@Bean
public Job job2(JobRepository jobRepository,
                 Flow validationFlow, Step exportStep) {
    return new JobBuilder("job2", jobRepository)
            .start(validationFlow)
            .next(exportStep)
            .end()
            .build();
}

validationFlow를 두 Job에서 재사용한다. 검증 로직이 변경되면 Flow만 수정하면 된다.

병렬 실행

독립적인 Step을 동시에 실행해서 처리 시간을 단축할 수 있다. Split을 사용한다.

@Bean
public Job parallelJob(JobRepository jobRepository,
                        Flow flow1, Flow flow2, Step finalStep) {
    return new JobBuilder("parallelJob", jobRepository)
            .start(new FlowBuilder<SimpleFlow>("splitFlow")
                    .split(new SimpleAsyncTaskExecutor())
                    .add(flow1, flow2)
                    .build())
            .next(finalStep)
            .end()
            .build();
}

flow1과 flow2가 별도 스레드에서 동시에 실행되고, 둘 다 완료되면 finalStep이 실행된다.

Decider — 프로그래밍 방식의 분기

ExitStatus 문자열 비교보다 복잡한 분기 로직이 필요할 때 JobExecutionDecider를 사용한다.

public class DayOfWeekDecider implements JobExecutionDecider {

    @Override
    public FlowExecutionStatus decide(JobExecution jobExecution,
                                       StepExecution stepExecution) {
        DayOfWeek day = LocalDate.now().getDayOfWeek();
        if (day == DayOfWeek.MONDAY) {
            return new FlowExecutionStatus("WEEKLY");
        }
        return new FlowExecutionStatus("DAILY");
    }
}

@Bean
public Job deciderJob(JobRepository jobRepository,
                       Step dailyStep, Step weeklyStep) {
    JobExecutionDecider decider = new DayOfWeekDecider();

    return new JobBuilder("deciderJob", jobRepository)
            .start(decider)
                .on("DAILY").to(dailyStep)
                .from(decider)
                .on("WEEKLY").to(weeklyStep)
            .end()
            .build();
}

월요일에는 주간 집계, 나머지 날에는 일일 집계를 실행하는 패턴이다.

자주 하는 실수