Spring Batch ( + Quartz )

배치

데이터를 일괄적으로 모아 처리하는 작업이다.
이때 데이터를 실시간으로 처리하지 않고 일정 시간에 일괄적으로 처리한다.
배치 애플리케이션은 다음 조건을 만족해야 한다.

배치 애플리케이션 조건

• 대용량 데이터 - 대량의 데이터를 가져오거나, 전달 및 계산할 수 있어야 한다.
• 자동화 - 외부 개입 없이 실행할 수 있어야 한다.
• 견고성 - 배치 애플리케이션 간 잘못된 데이터를 충돌/중단 없이 처리할 수 있어야 한다.
• 신뢰성 - 로깅 및 알림으로 잘못된 부분을 추적할 수 있어야 한다.
• 성능 - 다른 애플리케이션을 방해하지 않고 지정한 시간 안에 처리할 수 있어야 한다.

Batch vs Quartz

배치가 실시간 처리가 아닌 특정 시간에 데이터를 일괄 처리한다는 점에서 스케줄링과 개념을 혼동하곤 하는데 결론부터 말하면 Spring Batch은 Quartz가 아니다!
둘은 사용 목적 자체가 다르다.

• Quartz : 매시간 지정한 시간에 지정된 동작을 수행하는 스케줄링 라이브러리
• Spring Batch : 대용량 데이터를 자동화 처리하는 프레임워크

배치를 효율적으로 사용하기 위해 스케줄링을 보완제로 사용하는 것이지 둘이 같은 개념이 아니다. 따라서 두 기능을 잘 조합해서 배치 애플리케이션을 만들어야 한다.

Spring Batch

스프링 배치는 배치 애플리케이션 사용을 용이하게 하는 스프링 기반 프레임워크이다. 로깅 및 추적, 트랜잭션 관리, 작업 처리 통계, 작업 재시작, 건너뛰기, 리소스 관리 등의 기능을 제공한다.

Spring Batch Architecture

스프링 배치는 잡(Job), 스텝(Step), 리더(Reader), 프로세서(Processor), 라이터(Writer) 등의 개념을 이용하여 배치 처리를 설계하고 실행한다.

• Job Launcher : Job을 실행하기위한 인터페이스
• Job : 배치 처리 과정을 하나의 단위로 만들어 놓은 엔티티 객체
  • Batch 작업(단위)로 여러개의 step으로 구성
• Step : 배치 Job의 배치처리를 정의하고 순차적인 단계를 캡슐화 한 도메인 객체
  • ItemReader, ItemProcessor, ItemWriter로 구성
• Item : 작업에 사용되는 데이터
  • batch에 정의한 작업을 데이터베이스에 넣을 때, 데이터베이스에 있는 하나의 row가 하나의 item
• ItemReader: Step에서 한 항목씩 조회하는 작업. 모든 항목이 소진된 경우 null을 반환.
• ItemProcessor: 비즈니스 처리를 담당한다. 항목이 유효하지 않다고 판단되는 경우 null을 반환.
  • 자동으로 Null을 반환해주기 때문에 Optional로 객체를 한번 더 감싸는 비용을 감수하지 않아도 됨.
• ItemWriter: 지정된 단위로 아이템(데이터)를 출력하는 작업.
• JobRepository : 모든 저장(Persistence)매커니즘을 담당.
  • JobLauncher, Job, Step 구현체에 CRUD 기능을 제공

용어가 많아서 당황했겠지만 Spring Batch는 Launcher나 Repository를 제공하기 때문에 개발자는 로직을 짜는데만 집중하면 된다! 겁먹지 마시게들..

🧑🏻‍🏫 한줄요약

launcher로 실행된 job은 2개 이상의 step으로 구성되고, step을 read, process, write 한 뒤 repostiory를 통해 db에 저장된다.

이 외에도 많은 개념이 있지만 이번 포스팅에서는 Spring Batch를 간략히 소개하는것이 목적이므로 step이 tasklet 구성된다는 개념까지만 알고 지금부터는 코드로 구현해보겠다.

스케줄링을 이용한 Batch 구현

10초마다 ----- hello batch! ----- 를 출력하는 배치 스케줄러이다.

BatchConfig

• Job, Step, Tasklet을 정의하고 빈으로 등록한다.

@Configuration
public class BatchConfig extends DefaultBatchConfiguration {

    @Bean
    public Job testJob(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws DuplicateJobException {
        // JobBuilder를 사용하여 "testJob"이라는 job 생성
        Job job = new JobBuilder("testJob", jobRepository)
                .start(testStep(jobRepository, transactionManager)) // testStep을 시작 스텝으로 설정
                .build();
        return job;
    }

    public Step testStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        // StepBuilder를 사용하여 "testStep"이라는 step 생성
        Step step = new StepBuilder("testStep", jobRepository)
                .tasklet(testTasklet(), transactionManager) // testTasklet을 tasklet으로 설정하고 트랜잭션 매니저 사용.
                .build();
        return step;
    }

    public Tasklet testTasklet() {
        // Tasklet 생성
        return ((contribution, chunkContext) -> {
            System.out.println("----- hello batch! -----"); 
            // 원하는 비즈니스 로직 작성
            return RepeatStatus.FINISHED; // 작업이 완료되었음을 나타냄
        });
    }
}

BatchScheduler

• 스케줄링을 통해 10초마다 runJob 메서드가 실행

@RequiredArgsConstructor 
@Component
public class BatchScheduler {

    private final JobLauncher jobLauncher;
    private final JobRegistry jobRegistry;

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor() {
        // JobRegistryBeanPostProcessor 빈 생성
        JobRegistryBeanPostProcessor jobProcessor = new JobRegistryBeanPostProcessor();
        jobProcessor.setJobRegistry(jobRegistry); // JobRegistry를 설정하여 Job을 동적으로 등록/제거
        return jobProcessor;
    }

    @Scheduled(cron = "0/10 * * * * *") // 10초마다 스케줄링하여 실행
    public void runJob() {
        String time = LocalDateTime.now().toString(); // 현재 시간을 문자열로 변환
        try {
            Job job = jobRegistry.getJob("testJob"); // jobRegistry에서 "testJob"을 가져옴
            JobParametersBuilder jobParam = new JobParametersBuilder().addString("time", time); // JobParametersBuilder에 현재 시간을 추가함
            jobLauncher.run(job, jobParam.toJobParameters()); // Job을 실행
        } catch (NoSuchJobException e) {
            throw new RuntimeException(e); 
        } catch (JobInstanceAlreadyCompleteException |
                 JobExecutionAlreadyRunningException |
                 JobParametersInvalidException |
                 JobRestartException e
        ) {
            throw new RuntimeException(e); 
        }
    }
}

BatchApplication

@SpringBootApplication 
@EnableScheduling // 스케줄링 기능을 활성화
public class BatchApplication {
    public static void main(String[] args) {
        SpringApplication.run(BatchApplication.class, args);
    }
}

BatchApplication이 실행되면 의도한대로 10초마다 -----hello batch!----- 문자열이 출력된다.

이번 포스팅에서는 tasklet 단위로 Spring Batch에 대해 간략히 다뤘다. 기회가 된다면 chunk 단위로 ItemReader에 Paging 처리를 하여 구현해보는 것을 추천한다. 또 Batch는 QA 하기 어렵기 때문에 단위 테스트를 이용해서 내부 작업을 검사하고 테스트 코드를 작성하는 것을 적극 추천한다.

정리

배치 애플리케이션은 대용량 데이터 처리를 위해 JobLauncher, Job, Step으로 구성되며, 스케줄링은 이를 보완한다. 스프링 배치로 비즈니스 로직에 집중할 수 있으며, 다양한 운영 방식을 비즈니스 로직에 맞게 선택해야 합니다.

---

출처
✔️ spring batch architecture image
✔️ 배치 스케줄러 코드