06. Backup Batch - 2

이번 포스트에서는 지난 시간에 이어서 간단한 동작을 처리하는 Tasklet을 추가해서 Job을 완성해보자.

 

Tasklet을 활용한 Step

 

Tasklet

이전 포스트에서 배웠던 Step은 ItemReader -> [ItemProcessor ->] ItemWriter의 과정을 거쳐 chunk 단위로 데이터를 처리한다. 하지만 배치 잡 시작 전에 특정 초기화를 진행한다든지, 메인 처리 이후 마무리 작업을 진행하는 과정에는 특별한 ItemReader, ItemWriter 등이 필요 없는 상황도 발생한다. 이런 경우 Tasklet을 사용하여 TaskletStep을 구성할 수 있다.

Tasklet은 execute()는 TaskletStep에 의해 반복적으로 호출되는데 RepeatStatus.FINISHED(작업 끝)을 반환하거나 예외가 발생하면 종료된다. 또한 execute() 호출은 독립적인 Transaction 안에서 실행되기 때문에 실패하면 rollback 된다.

@FunctionalInterface
public interface Tasklet {
	@Nullable RepeatStatus execute(StepContribution contribution, 
                                   ChunkContext chunkContext) throws Exception;
}

파라미터로 전달되는 StepContribution 의 경우는 "이번 Step 실행에서 뭘 얼마나 했는지에 대한 결과 보고서"로써 처리 현황을 기록하고 ChunkContext는 "지금 내가 일하는 배치 청크 작업 현장의 상황 정보"로 실행 환경을 확인하는데 사용될 수 있다.

 

Tasklet을 활용한 이전 Step 결과 모니터링

Tasklet을 사용하기 위해서는 StepBuilder의 tasklet 메서드에 Tasklet 객체를 구성하면 된다.

@Bean
Step reportStep(JobRepository jobRepository,
    @Qualifier("quietjunTransactionManager") PlatformTransactionManager transactionManager,
    @Qualifier("quietjunDataSource") javax.sql.DataSource quietjunDataSource) {

    return new StepBuilder("reportStep", jobRepository)
        .tasklet((contribution, chunkContext) -> {
            String yearMonth = (String) chunkContext.getStepContext()
                                            .getJobParameters().get("yearMonth");
                    
            Collection<StepExecution> stepExecutions = 
                        chunkContext.getStepContext() // 현재 StepContext에서 출
                        .getStepExecution()           // StepExecution 을 거쳐
                        .getJobExecution()            // JobExecution 으로 이동
                        .getStepExecutions();         // 모든 StepExecution 들 중에서 필요 정보 얻기
                    
            long readCount = stepExecutions.stream()
                    .filter(stepExecution -> "backupPaymentStep".equals(stepExecution.getStepName()))
                    .findFirst()
                    .map(se -> se.getReadCount())
                    .orElse(0L);

            long writeCount = stepExecutions.stream()
                    .filter(se -> "backupPaymentStep".equals(se.getStepName()))
                    .findFirst()
                    .map(se -> se.getWriteCount())
                    .orElse(0L);

            // 실제 백업된 데이터 확인
            JdbcTemplate jdbcTemplate = new JdbcTemplate(quietjunDataSource);
            Integer backupCount = jdbcTemplate.queryForObject(
                    "SELECT COUNT(*) FROM quietjun.payment_backup WHERE payment_ym = ?",
                    Integer.class, yearMonth);

            String separator = "========================================";
            String message = """
                            %s
                            Payment 백업 배치 완료!
                            %s
                            대상 월: %s
                            읽은 건수: %d건
                            저장 건수: %d건
                            백업 확인: %d건
                            제외 건수: %d건 (paymentId null)
                            %s
                              """
            .formatted(separator, separator, yearMonth, readCount, writeCount, 
                            backupCount, (readCount - writeCount), separator);

            // 실무에서는 여기서 이메일 발송, 슬랙 알림 등
            log.debug(message);
            return RepeatStatus.FINISHED;
        }, transactionManager)
        .build();
}

 

Job 구성

 

기본

Job을 구성할 때는 JobBuilder를 사용한다. JobBuilder는 JobRepository를 필요로 하며 여러 개의 Step을 포함할 수 있다. step을 등록할 때는 start 이후 next 를 이용한다. 다음은 3개의 step으로 구성된 Job이다.

@Bean
public Job footballJob(JobRepository jobRepository) {
    return new JobBuilder("footballJob", jobRepository)
                     .start(playerLoad())         // step 1
                     .next(gameLoad())            // step 2
                     .next(playerSummarization()) // step 3
                     .build();
}

 

작업 실행 가로채기

Job의 생명 주기(작업 시작, 종료)에서 생기는 다양한 이벤트들을 통지받으면 유용한 경우가 있다. (예를 들면 작업 완료 보고를 메일로 보낸다든지.) 이를 위해 JobExecutionListener를 활용한다.

public interface JobExecutionListener {

    void beforeJob(JobExecution jobExecution);

    void afterJob(JobExecution jobExecution);
}

JobExecutionListener의 구현체를 Job에 등록해주면 job의 시작 전과 종료 후 이벤트를 처리할 수 있다.

@Bean
public Job footballJob(JobRepository jobRepository) {
    return new JobBuilder("footballJob", jobRepository)
                     .listener(sampleListener())
                     ...
                     .build();
}

참고로 afterJob의 경우 성공 여부에 상관 없이 호출되므로 연관 동작을 처리하기 위해서는 jobExecution의 status를 활용한다.

public void afterJob(JobExecution jobExecution){
    if (jobExecution.getStatus() == BatchStatus.COMPLETED ) {
        //job success
    }
    else if (jobExecution.getStatus() == BatchStatus.FAILED) {
        //job failure
    }
}

 

Job 구현

이제 앞서 생성한 backupPaymentStep과 repostStep을 이어서 paymentBackupJob을 만들어보자.

@Bean
Job paymentBackupJob(
            JobRepository jobRepository,
            Step backupPaymentStep,
            Step reportStep) {

    return new JobBuilder("paymentBackupJob", jobRepository)
                .start(backupPaymentStep) // 1. 백업
                .next(reportStep)         // 2. 보고
                .build();
}