Spring Batch

지난 포스팅에서 스케줄러에 대해서 다뤄봤습니다.

 

오늘은 배치 코드에 대해서 다뤄보려고 합니다.

 

build.gradle 설정

    implementation 'org.springframework.boot:spring-boot-starter-batch'

 

application.yml

spring:
  batch:
    jdbc:
      initialize-schema: always

 

기존에는 이 방식으로 메타데이터를 관리하는 스키마를 생성할 수 있었으나, 

Spring 3.X.X 버전으로 전환 후에 위 설정을 했음에도 불구하고 메타 테이블이 생성되지 않는 문제가 있었습니다.

 

그래서 3버전대를 사용하시는 분들은 아래와 같은 방법으로 테이블을 생성하시는걸 추천드립니다.

 

1. External Libraries -> Gradle: org.springframework.batch:spring-batch-core:5.1.2 -> org -> schema-[사용하는DB].sql

2. sql파일 안에 있는 DDL 실행

 

Post Entity

간단한 실습을 위해 아래와 같이 엔티티를 설계합니다.

@Entity
@Getter
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class Post {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long idx;

    private String title;
    private String content;
    @Setter
    private boolean published;  // 게시 여부
}

 

PostRepository

@Repository
public interface PostRepository extends JpaRepository<Post, Long> {
}

 

BatchConfig 파일 만들기

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    private final EntityManagerFactory entityManagerFactory;
    private final PostRepository postRepository;
    private final JobRepository jobRepository;
    private final PlatformTransactionManager transactionManager;

    @Autowired
    public BatchConfig(EntityManagerFactory entityManagerFactory, PostRepository postRepository,
                       JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        this.entityManagerFactory = entityManagerFactory;
        this.postRepository = postRepository;
        this.jobRepository = jobRepository;
        this.transactionManager = transactionManager;
    }

    @Bean
    public Job postJob() {
        return new JobBuilder("postJob", jobRepository)  // JobRepository를 사용하여 Job 생성
                .start(processPostStep())  // 단일 Step으로 구성된 Job
                .build();
    }

    // 단일 배치 Step 정의
    @Bean
    public Step processPostStep() {
        return new StepBuilder("processPostStep", jobRepository)  // JobRepository를 사용하여 Step 생성
                .<Post, Post>chunk(10, transactionManager)  // chunk 단위로 10개의 데이터를 처리, transactionManager 포함
                .reader(postReader())   // 데이터 읽기
                .processor(postProcessor())  // 데이터 처리
                .writer(postWriter())  // 데이터 저장
                .build();
    }

    // 미게시된 게시글을 DB에서 읽어오기
    @Bean
    public JpaPagingItemReader<Post> postReader() {
        System.out.println("reader action");
        return new JpaPagingItemReaderBuilder<Post>()
                .name("postReader")
                .entityManagerFactory(entityManagerFactory)
                .queryString("SELECT p FROM Post p WHERE p.published = false")  // 미게시된 게시글 조회
                .pageSize(3)
                .build();
    }

    // 게시글을 게시 처리
    @Bean
    public ItemProcessor<Post, Post> postProcessor() {
        return post -> {
            post.setPublished(true);  // 게시 상태로 변경
            return post;
        };
    }

    // 게시글을 저장
    @Bean
    public ItemWriter<Post> postWriter() {
        return posts -> postRepository.saveAll(posts);
    }
}

 

일단 시나리오는 다음과 같습니다.

1. 게시가된 게시글과, 게시가 되지 않은 게시글이 있습니다.

2. 페이징과, 청크를 활용해서 일정 개수의 게시글을 가져오고, 청크사이즈대로 게시가 되지 않은 게시글을 게시처리합니다.

3. 후에 게시 처리한 내용을 반영시킵니다.

 

- postReader

postReader에서는 DB에서 값을 읽어오는 과정입니다.

다양한 쿼리와, 페이지 처리를 통해 데이터를 읽어오면 됩니다.

 

- postProcessor

postProcessor에서는 데이터를 처리하는 과정을 진행합니다.

저는 시나리오대로, 게시글을 게시처리하기 위해서 setPublished true처리를 진행했습니다.

 

- postWriter

postWriter에서는 처리된 데이터를 다시 DB에 반영하는 역할을 합니다.

BatchScheduler

@Component
@RequiredArgsConstructor
public class BatchScheduler {

    private final JobLauncher jobLauncher;
    private final Job postJob;


    // 매시간 배치 작업을 실행하는 스케줄러 (cron 표현식으로 설정 가능)
    @Scheduled(cron = "30 * * * * ?")  //
    public void runBatchJob() {
        try {
            JobParameters params = new JobParametersBuilder()
                    .addLong("time", System.currentTimeMillis())  // 매번 새로운 파라미터로 실행
                    .toJobParameters();
            jobLauncher.run(postJob, params);  // 배치 작업 실행
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

 

---

이렇게 코드를 실행하고 나면, 주기적으로 게시글의 상태를 확인하면서 게시되지 않은 게시글을 게시처리를 하게되는 배치 시스템이 완성됩니다.

 

지금은 한 프로젝트 내에서 스케줄러와 배치시스템을 둘다 구축한 간단한 실습이지만,

 

다음에는 서버에 배치처리에 관련된 jar파일을 추출해서 셋팅하고, crontab 설정을 통해서 주기적으로 배치를 실행하고, 작업이 끝나면 꺼지게 하는 과정에 대해서도 다뤄보려고 합니다.

 

 

실행하는 과정에서 들었던 의문점 중 하나는 상상 이상으로 쿼리가 많이 날라가는 모습을 볼 수 있었습니다.

이 부분에 대해서는 좀 더 자세하게 알아보고 다뤄보는 시간을 가져보겠습니다.