Spring Batch 활용하기

 

 

 

Spring Batch

 

 

 

순서

Spring Boot에서 Spring Batch는 다음의 작업을 통해 사용할 수 있습니다.

• 의존성 추가
• DB에 Batch를 위한 테이블들 생성
• Config 클래스 작성
• JobLauncher로 Job 실행 

 

하나씩 알아보겠습니다. 

 

 

 

의존성 추가

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

build.gradle에 의존성을 추가하고 코끼리 표시를 눌러 라이브러리들을 다운받습니다.

 

 

 

DB에 Batch를 위한 테이블들 생성

Spring Batch를 실행시키기 위해 Spring Batch를 관리하기 위해 사용하는 테이블들이 Spring Boot에 연결된 DB에 있어야 합니다. 

이 테이블들의 SQL은 다운받은 라이브러리 안에 있습니다. 

Spring Boot에 연결된 DB에 맞는 SQL을 선택해 DB에 실행시켜 테이블들을 만듭니다. 

 

경로 : 

spring-batch-core 검색

→ org.springframework.batch.core

→ schema-mysql.sql

→ schema-mysql.sql을 mysql에 실행 

 

경로

 

Batch SQL 경로1

 

Batch SQL 경로2

 

 

 

Config 클래스 작성

다음은 Batch Config 클래스를 작성합니다. 

이곳에서는 Spring Batch로서 작업할 코드를 작성합니다. 

Spring Batch는 Job과 Step으로 이루어져 있습니다. 

Step은 동작을 수행하는 로직이 들어있습니다.

Job은 step들로 이루어져 있으며 Job을 실행시키면 step들이 차례로 수행합니다.

즉, 사용자들은 step에 로직을 짜고 Job에 step들을 배치시켜 원하는 동작을 수행하게 합니다. 

 

이 step은 Tasklet, Chunk 라는 2가지 방식이 있습니다. 

 

Tasklet 방식은 단일 작업을 수행하는데 사용됩니다. 

리소스를 초기화하거나 특정 상태를 체크하거나 DB를 클린업 하는 것과 같은 단일 작업을 수행할 때 사용됩니다. 

 

Chunk 방식은 데이터 항목들을 읽고, 처리하고, 쓰는 반복적인 작업을 수행할 때 사용됩니다.

대부분의 Batch에서는 Chunk 방식을 사용합니다. 

Chunk 방식은 ItemReader, ItemProcessor, ItemWriter 이 세 부분으로 구성됩니다. 

• ItemReader : 데이터를 한 항목 혹은 일관적으로 읽는 부분
• ItemProcessor : 각 항목에 대한 로직을 처리하는 부분
• ItemWriter : 처리된 항목들을 저장하는 부분 

Chunk 방식에는 chunk size를 정의하여 한 번에 몇 개의 항목들을 가져올지 정할 수 있습니다. 

 

 

Tasklet 예시 

package com.dotd.user.batch;


import lombok.RequiredArgsConstructor;
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;




@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
@Slf4j
public class BatchExampleConfig {
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    @Bean
    public Job simpleJob() {
        return jobBuilderFactory.get("simpleJob")
                .start(simpleStep1())
                .build();
    }

    @Bean
    public Step simpleStep1() {
        return stepBuilderFactory.get("simpleStep1")
                .tasklet( ((contribution, chunkContext) -> {
                    log.info(">>> simpleStep1");
                    return RepeatStatus.FINISHED;
                }))
                .build();
    }
}

 

 

 

Chunk 예시(JPA 사용)

package com.dotd.user.batch;


import com.dotd.user.entity.User;
import com.dotd.user.repository.UserRepository;
import lombok.RequiredArgsConstructor;
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.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.JpaItemWriter;
import org.springframework.batch.item.database.JpaPagingItemReader;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.persistence.EntityManagerFactory;
import javax.sql.DataSource;

@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
@Slf4j
public class BatchConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    private final EntityManagerFactory entityManagerFactory;

    @Bean
    public JpaPagingItemReader<User> jpaPagingItemReader() {
        JpaPagingItemReader<User> reader = new JpaPagingItemReader<>();
        reader.setEntityManagerFactory(entityManagerFactory);

        // JPQL 쿼리 -> MySQL이 아닌 Entitiy 이름이어야 한다.
        // User 의 모든 인스턴스를 선택
        // setQueryString은 해당 Reader 사용할 JPQL 쿼리를 설정
        // 데이터를 페이징 방식으로 읽어옴
        // User의 엔티티의 모든 인스턴스를 페이지 단위로 읽어옴
        reader.setQueryString("select u from User u");


        reader.setPageSize(100);
        return reader;
    }


    @Bean
    public ItemProcessor<User, User> userProcessor() {
        return user -> {

            // 유저 등급을 업데이트 하는 로직

            if(user.getUsedMoney() <= 100) {
                user.setTier("Bronze");
            }
            else if(user.getUsedMoney() <= 500) {
                user.setTier("Silver");
            }
            else if(user.getUsedMoney() <= 1000) {
                user.setTier("Gold");
            }

            // user 데이터를 가공하는 로직
            return user;
        };
    }

    @Bean
    public JpaItemWriter<User> jpaItemWriter() {
        JpaItemWriter<User> writer = new JpaItemWriter<>();
        writer.setEntityManagerFactory(entityManagerFactory);
        return writer;
    }


    @Bean
    public Step userStep() {
        return stepBuilderFactory.get("userStep")
                .<User, User>chunk(100)
                .reader(jpaPagingItemReader())
                .processor(userProcessor())
                .writer(jpaItemWriter())
                .build();
    }

    @Bean
    public Job userJob(Step userStep) {
        return jobBuilderFactory.get("userJob")
                .start(userStep)
                .build();
    }
}

 

 

 

JobLauncher로 Job 실행 

Config에서 만들어진 Job은 Bean으로 등록됩니다.

JobLauncher는 이렇게 등록된 Job을 불러와 실행시킵니다.

이 때 JobParameter 와 함께 실행시킵니다. 

JobParameter와 함께 실행하는 이유는 Spring Batch 작업의 인스턴스화와 실행을 구별하기 위해서 입니다.

Job을 실행하면 JobInstnace가 생성되는데 JobInstance는 Job의 이름과 JobParameter의 이름을 조합하여 이름을 정하여 유일성을 결정합니다. 

대부분 Job이 실행되는 시간으로 JobParameter를 사용합니다. 

동일한 파라미터와 이름으로 Job을 실행하면 Spring Batch는 이미 실행되었는지 확인하여 중복 실행을 방지합니다.

또한 만약 작업이 실패하면 다시 시작하는 기능을 제공하는데 동일한 파라미터를 사용해 재시작 할 수 있습니다.

JobParameter는 Spring Batch  로직 실행에 직접적인 영향을 주지 않습니다. 

 

 

 

JobLauncher로 Job 실행 (Rest API)

package com.dotd.user.controller;


import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequiredArgsConstructor
@RequestMapping("/batch")
@Slf4j
public class BatchController {

    private final JobLauncher jobLauncher;
    private final Job simpleJob;
    private final Job userJob;


    // user 등급 업데이트 메소드
    @PostMapping("/set-tier")
    public ResponseEntity<String> setTier() {
        try {
            /*
            jobParameters를 사용하는 이유는 Spring  Batch 작업의 인스턴스화와 실행을 구별하기 위해서다
            JobInstance는 job의 이름과 JobParameters의 이름을 조합하여 유일성이 결정된다.
            동일한 파라미터와 이름으로 Job을 실행하면 Spring Batch는 이미 실행되는지 확인하고 중복 실행을 방지한다.
            또, 만약 작업이 실패하면 다시 시작하는 기능을 제공하는데 동일한 파라미터를 사용해 재시작할 수 있다.
            parameter는 실행에 직접적인 영향을 주지 않습니다. 메타데이터 관리, 재시작 기능, JobInstance 식별 기능 등을 수행합니다.
             */

            JobParameters jobParameters = new JobParametersBuilder()
                    .addLong("time", System.currentTimeMillis())
                    .toJobParameters();
            jobLauncher.run(userJob, jobParameters);
            return ResponseEntity.ok("Batch job has been started.");
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                    .body("Failed to start the batch job.");
        }
    }


    // batch 실행 테스트
    @PostMapping("/start")
    public ResponseEntity<String> startBatch() {

        try {
            JobParameters jobParameters = new JobParametersBuilder()
                    .addLong("time", System.currentTimeMillis())
                    .toJobParameters();
            jobLauncher.run(simpleJob, jobParameters);
            return ResponseEntity.ok("Batch job has been started.");
        } catch (Exception e) {
            return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                    .body("Failed to start the batch job.");
        }
    }


}