[스프링] batch + scheduler로 주기적인 파일 삭제 구현

Spring에서 batch 와 scheduler를 사용하여
매일 같은시각에 업데이트된 시각이 일주일 이전인 파일을
서버와 DB에서 삭제하는 기능을 구현하였다.

1. batch, scheduler 개념

batch 란?

배치작업은 데이터를 실시간으로 처리하는게 아니라, 일괄적으로 모아서 한번에 처리하는 작업을 의미한다. 가령, 은행의 정산작업의 경우 배치 작업을 통해 일괄처리를 수행하게 되며 사용자에게 빠른 응답이 필요하지 않은 서비스에 적용할 수 있다. 특정 시간이후에는 자원을 거의 소비하지 않는 것이 특징이다.

 

Spring Batch

Spring Batch 는 로깅/추적, 트랜잭션 관리, 작업 처리 통계, 작업  재시작, 건너뛰기 등 대용량 레코드 처리에 필수적인 기능을 제공합니다. 또한 최적화 및 파티셔닝 기술을 통해 대용량 및 고성능 배치 작업을 가능하게 하는 고급 기술 서비스 및 기능을 제공합니다.

Spring Batch에서 배치가 실패하여 작업 재시작을 하게 된다면 처음부터가 아닌 실패한 지점부터 실행을 하게 됩니다.

또한 중복 실행을 막기 위해 성공한 이력이 있는 Batch는 동일한 Parameters로 실행 시 Exception이 발생하게 됩니다.

 

Spring Batch vs Quartz? Scheduler?

Spring Batch는 Scheduler가 아니기에 비교 대상이 아닙니다.

Spring Batch는 Batch Job을 관리하지만 Job을 구동하거나 실행시키는 기능은 지원하고 있지 않습니다. Spring에서 Batch Job을 실행시키기 위해서는 Quartz, Scheduler, Jenkins등 전용 Scheduler를 사용하여야 합니다.

 

---

Spring Batch 용어

Job

Job은 배치처리 과정을 하나의 단위로 만들어 놓은 객체입니다. 또한 배치처리 과정에 있어 전체 계층 최상단에 위치하고 있습니다.

JobInstance

JobInstance는 Job의 실행의 단위를 나타냅니다. Job을 실행시키게 되면 하나의 JobInstance가 생성되게 됩니다. 예를들어 1월 1일 실행, 1월 2일 실행을 하게 되면 각각의 JobInstance가 생성되며 1월 1일 실행한 JobInstance가 실패하여 다시 실행을 시키더라도 이 JobInstance는 1월 1일에 대한 데이터만 처리하게 됩니다.

JobParameters

JobInstance는 Job의 실행 단위라고 했습니다. 그렇다면 JonInstance는 어떻게구별 할까요? 이는 바로 JobParameters 객체로 구분하게 됩니다. JobParameters는 JobInstance 구별 외에도 개발자 JobInstacne에 전달되는 매개변수 역할도 하고 있습니다.

또한 JobParameters는 String, Double, Long, Date 4가지 형식만을 지원하고 있습니다.

JobExecution

JobExecution은 JobInstance에 대한 실행 시도에 대한 객체입니다. 1월 1일에 실행한 JobInstacne가 실패하여 재실행을 하여도 동일한 JobInstance를 실행시키지만 이 2번에 실행에 대한 JobExecution은 개별로 생기게 됩니다. JobExecution는 이러한 JobInstance 실행에 대한 상태,시작시간, 종료시간, 생성시간 등의 정보를 담고 있습니다.

Step

Step은 Job의 배치처리를 정의하고 순차적인 단계를 캡슐화 합니다. Job은 최소한 1개 이상의 Step을 가져야 하며 Job의 실제 일괄 처리를 제어하는 모든 정보가 들어있습니다.

StepExecution

StepExecution은 JobExecution과 동일하게 Step 실행 시도에 대한 객체를 나타냅니다. 하지만 Job이 여러개의 Step으로 구성되어 있을 경우 이전 단계의 Step이 실패하게 되면 다음 단계가 실행되지 않음으로 실패 이후 StepExecution은 생성되지 않습니다. StepExecution 또한 JobExecution과 동일하게 실제 시작이 될 때만 생성됩니다. StepExecution에는 JobExecution에 저장되는 정보 외에 read 수, write 수, commit 수, skip 수 등의 정보들도 저장이 됩니다.

ExecutionContext

ExecutionContext란 Job에서 데이터를 공유 할 수 있는 데이터 저장소입니다. Spring Batch에서 제공하느 ExecutionContext는 JobExecutionContext, StepExecutionContext 2가지 종류가 있으나 이 두가지는 지정되는 범위가 다릅니다. JobExecutionContext의 경우 Commit 시점에 저장되는 반면 StepExecutionContext는 실행 사이에 저장이 되게 됩니다. ExecutionContext를 통해 Step간 Data 공유가 가능하며 Job 실패시 ExecutionContext를 통한 마지막 실행 값을 재구성 할 수 있습니다.

JobRepository

JobRepository는 위에서 말한 모든 배치 처리 정보를 담고있는 매커니즘입니다. Job이 실행되게 되면 JobRepository에 JobExecution과 StepExecution을 생성하게 되며 JobRepository에서 Execution 정보들을 저장하고 조회하며 사용하게 됩니다.

JobLauncher

JobLauncher는 Job과 JobParameters를 사용하여 Job을 실행하는 객체입니다.

ItemReader

ItemReader는 Step에서 Item을 읽어오는 인터페이스입니다. ItemReader에 대한 다양한 인터페이스가 존재하며 다양한 방법으로 Item을 읽어 올 수 있습니다.

ItemWriter

ItemWriter는 처리 된 Data를 Writer 할 때 사용한다. Writer는 처리 결과물에 따라 Insert가 될 수도 Update가 될 수도 Queue를 사용한다면 Send가 될 수도 있다. Writer 또한 Read와 동일하게 다양한 인터페이스가 존재한다. Writer는 기본적으로 Item을 Chunk로 묶어 처리하고 있습니다.

ItemProcessor

Item Processor는 Reader에서 읽어온 Item을 데이터를 처리하는 역할을 하고 있다. Processor는 배치를 처리하는데 필수 요소는 아니며 Reader, Writer, Processor 처리를 분리하여 각각의 역할을 명확하게 구분하고 있습니다.

 

---

Spring Batch 사용하기

Spring Batch에서의 Job은 여러가지 Step의 모음으로 구성되어 있으며 Job은 순차적인 Step을 수행하며 Batch를 수행하게 됩니다. Step은 Tasklet 처리 방식과 Chunk 지향 처리 방식을 지원하고 있습니다.

 

 

이 이상은 다음 참고 링크에 가면 더 자세하게 나와 있습니다. 

Spring Batch란? 이해하고 사용하기(예제소스 포함) :: 히진쓰의 서버사이드 기술 블로그 (tistory.com)

Spring Batch란? 이해하고 사용하기(예제소스 포함)

[스프링/Spring] Batch 구조와 구성 요소 | Deeplify

[스프링/Spring] Batch 구조와 구성 요소

---

2. batch + scheduler 구현

내가 구현하려는 '주기적인 파일 삭제 구현'은 여러개의 step을 필요로 하지 않으므로 단일 step으로만 구성하였다.

1) build.gradle에 다음 코드를 추가한다.

> build.gradle

implementation("org.springframework.boot:spring-boot-starter-batch")
implementation("org.springframework.boot:spring-boot-starter-quartz")

testImplementation 'org.springframework.batch:spring-batch-test'

2) 아래 두줄을 application.properties에 추가한다.

#spring boot batch + scheduler
spring.batch.initialize-schema: always		# batch 스키마 자동 생성
spring.batch.job.enabled=false			# 시작과 동시에 실행되는건 방지

3) intellij 기준으로 'File>Invalidate Caches..' 로 캐시를 지우고 재시작시킨다.

4) main 함수가 있는 클래스에 @EnableScheduling과 @EnableBatchProcessing을 붙여준다.

> MeetingDocumentApplication.java

package com.tmax.meeting.document;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@EnableScheduling
@EnableBatchProcessing
@SpringBootApplication
public class MeetingDocumentApplication {

  public static void main(String[] args) {
    SpringApplication.run(MeetingDocumentApplication.class, args);
  }

}

5) batch 파일과 scheduler파일을 만든다.

6) batch 파일에서 수행하고자 하는 Job을 step으로 구성한다.

> config / BatchConfig.java

package com.tmax.meeting.document.config;

import com.tmax.meeting.document.model.Document;
import com.tmax.meeting.document.repository.DocumentRepository;
import com.tmax.meeting.document.service.DocService;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.time.LocalDateTime;

@Slf4j
@Configuration
@EnableBatchProcessing
public class BatchConfig {

  @Autowired
  public JobBuilderFactory jobBuilderFactory;

  @Autowired
  public StepBuilderFactory stepBuilderFactory;

  @Autowired
  private DocumentRepository documentRepository;

  @Autowired
  private DocService docService;

  @Bean
  public Job job() {

    Job job = jobBuilderFactory.get("job")
        .start(step())
        .build();

    return job;
  }

  @Bean
  public Step step() {
    return stepBuilderFactory.get("step")
        .tasklet((contribution, chunkContext) -> {
          log.info("Step!");
          // 업데이트 시각이 일주일 이전인 문서 목록을 가져옴
          // 1. 네이티브 쿼리 사용
          List<Document> limitedDocuments = documentRepository.selectLimitedDocument();
          // 2. JPQL 사용
          // LocalDateTime now = LocalDateTime.now();
          // LocalDateTime aWeekAgo = now.minusDays(7);
          // List<Document> limitedDocuments = documentRepository.findByUpdateAtLessThan(aWeekAgo)

          if (limitedDocuments.size() > 0 && limitedDocuments != null) {
            for (Document document : limitedDocuments) {
              // deleteDocument는 document_id를 받아 서버와 db에서 문서 삭제를 구현하는 서비스
              docService.deleteDocument(document.getDocumentId());
            }
          }
          return RepeatStatus.FINISHED;
        })
        .build();
  }
}

업데이트 시각이 일주일 이전인 문서 목록을 가져오는 쿼리문은 네이티브 쿼리, JPA로 구현해보았는데

1. 네이티브 쿼리 사용
> respository /DocumentRespository.java

@Repository
public interface DocumentRepository extends JpaRepository<Document, Long>{
  @Query(value = "SELECT * FROM document WHERE DATE(updated_at) < DATE_SUB(NOW(), INTERVAL 7 DAY)", nativeQuery = true)
  List<Document> selectLimitedDocument();
}

JPA로 작성하려고 보니 날짜계산이 들어가서 어떻게 할까 하다가 Querydsl 쓰기 귀찮아서... ㅎ 네이티브쿼리로 한것이다.
날짜 계산에 DATE_SUB( )을 사용한다는거!

2. JPQL 사용
JPQL로 하려면 LocalDateTime을 사용해서 7일전 날짜를 변수로 하여 이보다 빠른 update_at 값을 가지는 것을 가져오게 한다.
> config / BatchConfig.java 중..

import java.time.LocalDateTime;

LocalDateTime now = LocalDateTime.now();
LocalDateTime aWeekAgo = now.minusDays(7);

List<Document> limitedDocuments = documentRepository.findByUpdatedAtLessThan(aWeekAgo)

이렇게 7일전 날짜변수를 넘기면..
> respository /DocumentRespository.java

import java.time.LocalDateTime;

@Repository
public interface DocumentRepository extends JpaRepository<Document, Long>{
  List<Document> findByUpdatedAtLessThan(LocalDateTime aWeekAgo);
}

위 처럼 받아서 jpa를 수행한다.

7) batch를 수행하기 위한 scheduler를 scheduler 파일에 구성한다. (매일 오전 9시에 실행되도록)

> config / BatchScheduler.java

package com.tmax.meeting.document.config;

import java.util.HashMap;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.JobParameter;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersInvalidException;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class BatchScheduler {

  @Autowired
  private JobLauncher jobLauncher;

  @Autowired
  private BatchConfig batchConfig;

  @Scheduled(cron = "0 0 9 * * *")
  public void runJob() {

    // job parameter 설정
    Map<String, JobParameter> confMap = new HashMap<>();
    confMap.put("time", new JobParameter(System.currentTimeMillis()));
    JobParameters jobParameters = new JobParameters(confMap);

    try {
      jobLauncher.run(batchConfig.job(), jobParameters);
    } catch (JobExecutionAlreadyRunningException | JobInstanceAlreadyCompleteException
        | JobParametersInvalidException | org.springframework.batch.core.repository.JobRestartException e) {

      log.error(e.getMessage());
    }
  }
}

위에서 jobLauncher.run()메소드는 첫번째 파라미터로 Job, 두번째 파라미터로 Job Parameter를 받고 있다.
Job Parameter의 역할은 반복해서 실행되는 Job의 유일한 ID이다.

0000-00-00 01:30:43.702  INFO 16963 --- [   scheduling-1] o.s.batch.core.step.AbstractStep         : Step: [tutorialStep] executed in 14ms
... 생략 ...
0000-00-00 01:30:48.748  INFO 16963 --- [   scheduling-1] o.s.batch.core.job.SimpleStepHandler     : Step already complete or not restartable, so no action to execute: StepExecution: id=1, version=3, name=tutorialStep, status=COMPLETED, exitStatus=COMPLETED, readCount=0, filterCount=0, writeCount=0 readSkipCount=0, writeSkipCount=0, processSkipCount=0, commitCount=1, rollbackCount=0, exitDescription=

위처럼, Job Parameter에 동일한 값이 세팅되도록 하면 두번째부터 실행되는 Job의 Step은 실행되지 않는다.