[spring batch] 스프링 배치 이해하고 적용하기

각종 공연을 예매하고 알림을 받을 수 있는 서비스를 만들고 있었다. 여기서 예매한 공연의 시작 하루 전 회원들에게 일괄적으로 알림을 보내주는 기능을 구현했다. 

카카오톡 알림을 보내려 했지만 사업자 번호가 있어야 한다고 해서 아쉬운데로 sms 전송으로 대체했다.

 

sms push 알림은 naver cloud에서 제공하는 sens를 사용했다. coolsms가 제일 보편적으로 많이 쓰이는 것 같던데 훨씬 저렴하게 사용할 수 있었다. (지금은 사업자만 가능하도록 네이버에서 막음.. 스펨문제)

 

아무튼 위 기능을 구현하려고 이것저것 찾아보다 스케줄러와 배치에 대해 알게 되었고 현재 진행중인 프로젝트에 스프링 배치를 적용시키기로 했다.

 

스프링 배치는 어떤 작업에 쓰일까?

예약 시간에 광곳어 메시지 발송, 결제 정산 작업, 운영을 위해 필요한 통계 데이터 구축, 대량 데이터를 필요로 하는 모델 학습 작업 등 특정 시간에 대량의 데이터를 처리해야 하는 경우에 많이 쓰였다. 

 

스프링 배치는 스케줄러와 혼동하는 경우가 많은데 전혀 다른 개념이다. 둘의 차이는 다른 글에서 다루겠다.

https://dgjinsu.tistory.com/20

 

 

 

Spring batch 용어 정리

 

• Job
  • 배치처리 과정을 하나의 단위로 만들어 놓은 객체.
  • 전체 배치 프로세스를 캡슐화하는 엔티티, 실행시킬 작업.

    

• JobInstance
  • 논리적인 Job 의 실행의 단위.
  • Job 을 실행시키면 하나의 JobInstance 가 생성.
    • 예를들어, 1월 1일 실행, 1월 2일 실행을 하게 되면 각각의 JobInstance 가 생성되고, 만약 1월 1일에 실행한 JobInstance 가 실패하여 이후 재실행 하더라도 이 JobInstance 는 1월 1일에 대한 데이터만 처리.
• JobParameters
  • Job 을 실행하는데 사용하는 파라미터의 집합으로, Job 이 실행되는 동안에 JobInstance 의 식별 외에도 매개변수 역할로 사용.
• JobExecution
  • JobInstance 에서 실행 시도에 대한 객체.
  • 실행에 대한 JobExecution 은 개별로 생성.
  • JobInstance 실행에 대한 상태, 시작시간, 종료시간, 생성시간 등의 정보를 가짐.
    • 예를들어, 1월 1일에 실행한 JobInstance 가 실패하여 재실행을 하여도 동일한 JobInstance 를 실행시키지만 이 2번에 실행에 대한 Job Execution 은 개별로 생성.
• Step
  • Job 의 배치처리를 정의하고 순차적인 단계를 캡슐화.
  • Job 은 최소 1개 이상의 Step 을 가져야 하고, Job 의 실제 일괄 처리를 제어하는 모든 정보를 포함.
  • Step 의 내용은 개발자의 선택에 따라 구성.
  • Tasklet 처리 방식과 Chunk 지향 처리 방식을 지원
    • Tasklet
      • 단순한 데이터 프로세스 처리에 적합한 모델.
      • SQL 1회 명령 등 단순하거나, 작업 프로세스의 표준화가 어려운 복잡한 경우에 Custom 작업 생성을 위해 사용.
    • Chunk
      • 메모리에 가지고 있기 너무 많고, 큰 데이터 들을 효율적으로 처리하는데 적합.
      • 일정양의 데이터를 일괄적으로 read / process / write 프로세스 흐름에 따라 표준화하여 작업을 구현.
      • 하나의 Transaction 안에서 처리할 Item 의 덩어리.
      • chunk size 가 10이라면 하나의 transaction 안에서 10개의 item 에 대해 처리 하고 commit.

        

• StepExecution
  • JobExceution 과 동일하게 Step 실행 시도에 대한 객체.
  • 실제 시작이 될 때 생성.
  • 시작 시간, 종료 시간, 상태, 종료 상태, Commit Count, ItemCount, Skip Count 등 실행에 대한 다양한 정보를 가짐.
    • 예를들어, Job 이 여러 단계의 Step 으로 구성되어 있을 경우 이전 단계의 Step 이 실패하게 되면 다음 단계가 실행되지 않음으로 실패 이후 StepExecution 은 생성되지 않음.
• ExecutionContext
  • Job 에서 데이터를 공유 할 수 있는 데이터 저장소.
  • Spring Batch 에서는 JobExecutionContext, StepExecutionContext 를 지원한다.
    • JobExecutionContext
      • Commit 시점에 저장
    • StepExecutionContext
      • 실행 사이에 저장 ExecutionContext 를 통해 Step 간 Data 공유가 가능하고, Job 실패시 ExecutionContext 를 통해 마지막 실행 값을 재구성 할 수 있음.
• JobRepository
  • 수행되는 Job 의 모든 정보를 담고 있는 저장소.
  • 어떠한 Job 이 언제 수행되었고, 언제 종료하고, 몇 번 실행되었고, 실행에 대한 결과가 어떤지 등의 Batch 수행과 과련된 모든 Meta Data 가 저장.
  • Job 이 실행되게 되면 JobRepository 에 JobExecution 과 StepExecution 을 생성하고, JobRepository 에서 Execution 정보들을 저장하고 조회하며 사용.

 

 

Meta-Table Schema

- Spring Batch 에서는 6개의 Meta Table 과 3개의 Sequence Table 이 존재한다.

- 작업이 수행될 때 마다 실행된 Job 에 대한 다양한 정보들을 저장하여 관리한다.

- 일반적으로는 Meta Table 이 없으면 Spring Batch 를 실행할 수 없다.

 

 

Spring Batch 5 변경점

 

Spring Boot 3(=Spring Framework 6)부터 Spring Batch 5 버전을 사용하게 업데이트 되었다. Batch 5에 변경점이 많이 생겨 기존의 4버전과 다른 부분이 많이 생겼다. 처음 배치를 구현해보는데 Spring Batch5 버전의 참고할 자료가 많이 없어 고생했다. 

 

내가 구현하면서 발견한 변경점은 아래와 같다.

 

1. JobBuilderFactory, StepBuilderFactory deprecated

// Sample with v4
@Configuration
@EnableBatchProcessing
public class MyJobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Bean
    public Job myJob(Step step) {
        return this.jobBuilderFactory.get("myJob")
                .start(step)
                .build();
    }
}
// Sample with v5
@Configuration
@EnableBatchProcessing
public class MyJobConfig {

    @Bean
    public Job myJob(JobRepository jobRepository, Step step) {
        return new JobBuilder("myJob", jobRepository)
                .start(step)
                .build();
    }
}

 

 

2. JobRepository, TransactionManager 명시적으로 변경

 

내부적으로 필요한 객체를 명시적으로 표시해 주는 방식으로 변경되었다

// Sample with v4
@Configuration
@EnableBatchProcessing
public class MyStepConfig {

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public Step myStep() {
        return this.stepBuilderFactory.get("myStep")
                .tasklet(..) // or .chunk()
                .build();
    }
}
// Sample with v5
@Configuration
@EnableBatchProcessing
public class MyStepConfig {

    @Bean
    public Tasklet myTasklet() {
       return new MyTasklet();
    }

    @Bean
    public Step myStep(JobRepository jobRepository, Tasklet myTasklet, PlatformTransactionManager transactionManager) {
        return new StepBuilder("myStep", jobRepository)
                .tasklet(myTasklet, transactionManager) // or .chunk(chunkSize, transactionManager)
                .build();
    }
}

 

 

3. JobParameter 지원 범위 확대

 

v4 에서의 스프링 배치는 Job parameter 로 Long, String, Date, Double 만 사용이 가능했다.

v5 에서는 여기에 더해 converter 를 직접 구현하는 것으로 모든 종류의 타입을 JobParameter 로 사용할 수 있도록 개선됐다. 

 

 

 

 

본격적으로 알림 일괄처리를 개발해보자

 

SmsJobConfiguration

@Slf4j
@Configuration
@RequiredArgsConstructor
public class SmsJobConfiguration {
    private final SmsService smsService;
    private final EntityManagerFactory entityManagerFactory;
    @Bean @Qualifier(value = "smsSendJob")
    public Job smsSendJob(JobRepository jobRepository, Step smsSendStep) {
        return new JobBuilder("smsSendJob", jobRepository)
                .start(smsSendStep)
                .build();
    }
    @Bean
    public Step smsSendStep(JobRepository jobRepository, PlatformTransactionManager platformTransactionManager){
        return new StepBuilder("smsSendStep", jobRepository)
                .<Ticketing, Ticketing>chunk(100, platformTransactionManager)
                .reader(itemReader())
                .processor(itemProcessor())
                .writer(itemWriter())
                .build();
    }

    @Bean
    public ItemReader<Ticketing> itemReader() {
        LocalDateTime tomorrow  = LocalDateTime.now().plusDays(1);

        // DateTimeFormatter를 사용하여 포맷 지정
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

        // 00:00로 설정
        LocalDateTime tomorrowStart = tomorrow.withHour(0).withMinute(0).withSecond(0).withNano(0);
        // 23:59로 설정
        LocalDateTime tomorrowEnd = tomorrow.withHour(23).withMinute(59);

        return new JpaCursorItemReaderBuilder<Ticketing>()
                .name("itemReader")
                .entityManagerFactory(entityManagerFactory)
                .queryString("select t from Ticketing t join fetch t.member m where t.event.startEvent between :tomorrowStart and :tomorrowEnd")
                .parameterValues(Map.of("tomorrowStart", tomorrowStart, "tomorrowEnd", tomorrowEnd))
                .build();
    }

    @Bean
    public ItemProcessor<Ticketing, Ticketing> itemProcessor() {
        return ticketing -> {
            Member member = ticketing.getMember();
            String eventName = ticketing.getEvent().getName();
            LocalDateTime startEvent = ticketing.getEvent().getStartEvent();

            int month = startEvent.getMonthValue(); // 월(month) 추출
            int day = startEvent.getDayOfMonth();   // 일(day) 추출
            int hour = startEvent.getHour();        // 시(hour) 추출
            int minute = startEvent.getMinute();    // 분(minute) 추출
            String content = month + "월" + day + "일 " + hour + "시" + minute + "분에 " + eventName + "이 예매되어 있습니다. 시간에 맞춰 방문해주시기 바랍니다.";
            log.info(member.getPhone() + " 해당 전화번호로 [" + content + "] 문자 발송되었습니다.");
            smsService.sendSms(member.getPhone(), content);
            return ticketing;
        };
    }

    @Bean
    public JpaItemWriter<Ticketing> itemWriter() {
        return new JpaItemWriterBuilder<Ticketing>()
                .entityManagerFactory(entityManagerFactory)
                .build();
    }
}

 

전체 코드는 위와 같다.

 

 

하나씩 뜯어서 보면

 

- Job을 빈으로 등록하며 Step을 Job에 등록

@Bean @Qualifier(value = "smsSendJob")
public Job smsSendJob(JobRepository jobRepository, Step smsSendStep) {
    return new JobBuilder("smsSendJob", jobRepository)
            .start(smsSendStep)
            .build();
}

 

 

- Step 정의 (reader, processor, writer 세팅)

@Bean
public Step smsSendStep(JobRepository jobRepository, PlatformTransactionManager platformTransactionManager){
    return new StepBuilder("smsSendStep", jobRepository)
            .<Ticketing, Ticketing>chunk(5, platformTransactionManager)
            .reader(itemReader())
            .processor(itemProcessor())
            .writer(itemWriter())
            .build();
}

 

 

- reader

- JpaCursorItemReaderBuilder를 사용하여 조건에 맞는 item 조회

@Bean
public ItemReader<Ticketing> itemReader() {
    LocalDateTime tomorrow  = LocalDateTime.now().plusDays(1);

    // DateTimeFormatter를 사용하여 포맷 지정
    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    // 00:00로 설정
    LocalDateTime tomorrowStart = tomorrow.withHour(0).withMinute(0).withSecond(0).withNano(0);
    // 23:59로 설정
    LocalDateTime tomorrowEnd = tomorrow.withHour(23).withMinute(59);

    return new JpaCursorItemReaderBuilder<Ticketing>()
            .name("itemReader")
            .entityManagerFactory(entityManagerFactory)
            .queryString("select t from Ticketing t join fetch t.member m where t.event.startEvent between :tomorrowStart and :tomorrowEnd")
            .parameterValues(Map.of("tomorrowStart", tomorrowStart, "tomorrowEnd", tomorrowEnd))
            .build();
}

 

 

- processor

- reader에서 읽어드린 item을 활용해 필요한 로직 수행.

- 문자를 보내는 로직이 여기에 들어간다

@Bean
    public ItemProcessor<Ticketing, Ticketing> itemProcessor() {
        return ticketing -> {
            Member member = ticketing.getMember();
            String eventName = ticketing.getEvent().getName();
            LocalDateTime startEvent = ticketing.getEvent().getStartEvent();

            int month = startEvent.getMonthValue(); // 월(month) 추출
            int day = startEvent.getDayOfMonth();   // 일(day) 추출
            int hour = startEvent.getHour();        // 시(hour) 추출
            int minute = startEvent.getMinute();    // 분(minute) 추출
            String content = month + "월" + day + "일 " + hour + "시" + minute + "분에 " + eventName + "이 예매되어 있습니다. 시간에 맞춰 방문해주시기 바랍니다.";
            log.info(member.getPhone() + " 해당 전화번호로 [" + content + "] 문자 발송되었습니다.");
            smsService.sendSms(member.getPhone(), content);
            return ticketing;
        };
    }

 

 

 

- writer

- 변경된 item들을 다시 저장하는 부분이다. 

- 나는 item들의 데이터를 따로 수정하지 않았기 때문에 저장할 건 없다.

@Bean
public JpaItemWriter<Ticketing> itemWriter() {
    return new JpaItemWriterBuilder<Ticketing>()
            .entityManagerFactory(entityManagerFactory)
            .build();
}

 

---

 

 

이렇게 완성한 Batch 로직을 Sheduler에 등록하여 원하는 시간대에 실행해주면 된다. 

 

ScheduledJob

    @Scheduled(cron = "0 0 20 * * *", zone = "Asia/Seoul") // 초 분 시 일 월 요일
    public void sendReminderMessages() {
        Map<String, JobParameter<?>> confMap = new HashMap<>();
        confMap.put("time", new JobParameter(System.currentTimeMillis(), Long.class));
        JobParameters jobParameters = new JobParameters(confMap);

        try {
            jobLauncher.run(SmsSendJob, jobParameters);
            jobLauncher.run(alarmJob, jobParameters);
        } catch (JobExecutionAlreadyRunningException e) {
            throw new RuntimeException(e);
        } catch (JobRestartException e) {
            throw new RuntimeException(e);
        } catch (JobInstanceAlreadyCompleteException e) {
            throw new RuntimeException(e);
        } catch (JobParametersInvalidException e) {
            throw new RuntimeException(e);
        }
        log.info("스케줄러 실행");
    }

 

JobParameter를 사용하는 이유 중 하나는 Spring Batch가 각 작업을 고유하게 식별할 수 있도록 하기 위해서이다.

Spring Batch는 각 작업 실행에 대한 유일한 키를 가져야 하며, 이러한 키는 JobParameters에 의해 제공된다.