Spring batch 5.0 간단한 구현

Spring Batch 5.0

---

제가 작성하는 글의 Spring Batch는 5.0 버전입니다.

• 변경점

• @EnableBatchProcessing
  • 5버전부터 더이상 해당 어노테이션이 필수가 아니게 변경되었습니다.
  • Batch와 관련된 Bean을 등록하게 해주는 필수 어노테이션이었지만 이젠 사용하지 않아도 됩니다.
• JobBuilderFactory, StepBuilderFactory deprecated
  • 두 빌더 팩토리 대신 JobBuilder와 StepBuilder를 사용하고 있습니다.
• JobRepository, TransactionManager 명시적
  • 두 용어를 명시적으로 작성하도록 변경되었습니다.

 

 

 

간단한 구현

---

Spring Batch 5.0버전에 대한 간단한 예제가 많이 없어서 GitHub에서 예제를 참고하여 본 글을 작성하게 되었습니다.

https://github.com/caligula95/spring-batch-example

GitHub - caligula95/spring-batch-example

해당 github에 대한 유튜브도 있으니 찾아보시면 좋을 것 같습니다.

 

Spring Batch 관련 용어는 따로 정리해두었습니다.

https://hollo-coding.tistory.com/8

Spring batch 알아보기

 

 

 

Spring Initializr

---

Spring initializr

IDE - Intelli J

DB - H2 DataBase

필요한 Dependencies는 Spring initalizr를 이용해 구현했습니다.

 

 

 

1. 간단한 Tasklet 구현

---

Job 생성

Configuration

· TestJob( JobRepository jobRepository, PlatformTransactionManager manager )

   - JobRepository : Job을 관리하고 저장하기 위한 JobRepository

   - PlatfromTransactionManager : 트랜잭션의 관리를 위한 Spring의 인터페이스

 

·  JobBuilder( "TestJob" , jobRepository )

   - JobBuilder( "이름", 저장소 ) : 해당 Job의 이름과, Job을 관리하고 저장하는 JobRepository를 설정합니다.

 

·  incrementer(new RunIncrementer())

    Job을 식별하기 위해  매번 새로운 JobInstance를 생성해줍니다.

   - incrementer() : JobParameterIncrementer는 JobBuilder에 incrementer() 메서드를 이용해서 설정해 줄 수 있다. 

 

·   start() : 실행할 Tasklet 또는 Chunk 

---

Tasklet 생성

 - log에 "hello"를 출력하는 간단한 Tasklet입니다.

Tasklet

 

- 처음 Class 생성 후 implements를 하면 다음과 같은 Method가 필요로 합니다.

 

execute 메소드를 구현 후 RepeatStatus 객체를 반환합니다.

RepeatStatus : 해당 클래스를 반복적으로 실행할 지 결정하는 Enum 타입입니다.

RepeatStatus.CONTINUABLE - 해당 Tasklet을 다시 실행한다
RepeatStatus.FINISHED - 처리의 성공 여부 관계없이 Tasklet을 완료하고 다음 처리를 이어서 한다

StepContribution : 아직 커밋되지 않은 현재 트랜잭션에 대한 정보

ChunkContext : Tasklet 내에서 처리 중인 Chunk와 관련된 정보

---

결과

실행 결과

 

 

 

2. 간단한 Chunk 구현

---

Entity와 Repository 생성

- Chunk를 만들기 전에 사용할 BookEntity와 BookRepository를 생성합니다.

 

---

SQL 설정

 - resources > import.sql을 생성해 다음의 코드를 입력합니다.

# import.sql

insert into book_table(title, author, year_of_publishing) values ('The Great Gatsby', 'F. Scott Fitzgerald', 1925);
insert into book_table(title, author, year_of_publishing) values ('To Kill a Mockingbird','Harper Lee',1960);
insert into book_table(title, author, year_of_publishing) values ('1984','George Orwell',1949);

 

 - Console에 실행이 잘되는지 확인하기 위해 다음의 코드를 application.properties에 입력합니다.

 

다음의 코드가 잘 작동되는지 확인하기 위해 간단한 Controller를 구현해 확인했습니다.

 

잘 작동되는지 확인이 되었으니 Chunk를 작성해보겠습니다.

 

---

Chunk 생성

 - Chunk에는 ItemReader, ItemProcessor, ItemWriter가 필요로 합니다.

 

1. ItemReader

 - ItemReader를 상속받아 read() 메소드를 재정의 합니다.

 

 

방금 전에 만든 url에서 Book 정보를 받아오기 위하여 다음과 같이 코드를 작성했습니다.

public class BookReader implements ItemReader<BookEntity> {

    private final String url;
    private final RestTemplate restTemplate;
    private int nextBook;
    private List<BookEntity> bookList;

    public BookReader(String url, RestTemplate restTemplate) {
        this.url = url;
        this.restTemplate = restTemplate;
    }

    // 외부의 정보를 List<BookEntity>형태로 반환
    private List<BookEntity> fetchBooks(){
        // GET 요청을 보내고 응답을 ResponseEntity로 받습니다.
        ResponseEntity<BookEntity[]> response = restTemplate.getForEntity(this.url, BookEntity[].class);
        //ResponseEntity에서 Book 정보를 배열로 얻어 옵니다.
        BookEntity[] books = response.getBody();

        if (books != null){
            return Arrays.asList(books);
        }
        return null;
    }

    // read 메소드 재정의
    @Override
    public BookEntity read() throws Exception{
        if (this.bookList == null){ bookList = fetchBooks(); }
        BookEntity bookEntity = null;

        if (nextBook < bookList.size()){
            bookEntity = bookList.get(nextBook);
            nextBook++;
        } else {
            nextBook = 0;
            bookList = null;
        }
        return bookEntity;
    }
}

 

 

2. ItemProcessor

 - BookTitleProcessor는 Book의 Title을 모두 대문자로 만드는 Processor입니다.

 - ItemProcessor는 < I, O >을 설정해줘야 합니다.

   I  Generic : ItemReader에서 받을 데이터 타입

   O  Generic : ItemWriter에서 보낼 데이터 타입

public class BookTitleProcessor implements ItemProcessor<BookEntity, BookEntity> {
    @Override
    public BookEntity process(BookEntity item) throws Exception {
        item.setTitle(item.getTitle().toUpperCase());
        return item;
    }
}

 

 

3. ItemWriter

 - bookRepository에 Chunk에 담긴 Book 정보를 모두 저장합니다.

public class BookWriter implements ItemWriter<BookEntity> {
    @Autowired
    private BookRepository bookRepository;

    @Override
    public void write(Chunk<? extends BookEntity> chunk) throws Exception {
        bookRepository.saveAll(chunk.getItems());
    }
}

 

 

4. Chunk

·  StepBuilder( "TestChunkStep" , jobRepository )

   - StepBuilder ( "이름", 저장소 ) : 해당 Step의 이름과, Job을 관리하고 저장하는 JobRepository를 설정합니다.

 

·  <BookEntity, BookEntity>chunk(10, manager)

   - 첫번째 BookEntity : reader에서 반환되는 데이터 타입 

   - 두번째 BookEntity : processor와 writer에서 사용되는 데이터 타입

   - chunk(10, manager)

      10 : 한 번에 처리되는 청크의 크기 ( 10개의 BookEntity이 한 번에 처리됩니다. )

       manager : 트랜잭션 관리자

 

 

5. ItemReader, ItemProcessor, ItemWriter

@Bean
// Step-scoped bean을 위한 편리한 어노테이션
// @Bean을 @StepScope로 표시하는 것은 @Scope(value="step"), proxyMode=TARGET_CLASS)로 표시하는 것과 같습니다.
@StepScope
public ItemReader<BookEntity> ChunkReader(){
    return new BookReader("http://localhost:8080/book/findall", new RestTemplate());
}

@Bean
@StepScope
public ItemProcessor<BookEntity, BookEntity> Chunkprocessor() {
     return new BookTitleProcessor();
}

@Bean
@StepScope
public ItemWriter<BookEntity> ChunkWriter() {
    return new BookWriter();
}

---

실행

 - 다음과 같이 Console 창이 실행됩니다.

Console

---

결과

- BookTitleProcessor에서 설정한대로 Title이 모두 대문자로 바뀐걸 확인할 수 있습니다.

 

 

 

3. 여러개의 Processor 사용하기

---

BookAuthorProcessor 추가

public class BookAuthorProcessor implements ItemProcessor<BookEntity, BookEntity> {

    @Override
    public BookEntity process(BookEntity item) throws Exception {
        item.setAuthor("By " + item.getAuthor());
        return item;
    }
}

 

---

ChunkProcessor 수정

 - CompositeItemProcessor를 이용해 여러개의 Processor 적용할 수 있습니다.

 - 현재 사용중인 BookAuthorProcessor와 BookTitleProcessor의 < I, O >값이 같기 때문에 CompositeItemProcessor의 뒤에

   <BookEntity, BookEntity>를 작성했습니다.

@Bean
    @StepScope
    public ItemProcessor<BookEntity, BookEntity> Chunkprocessor() {
        CompositeItemProcessor<BookEntity, BookEntity> process = new CompositeItemProcessor<>();
        process.setDelegates(List.of(new BookAuthorProcessor(), new BookTitleProcessor()));
        return process;
    }

---

결과

 

 

 

4. 여러개의 Step 사용하기

---

- 전에 작성해두었던 Job에 .start()와 next()를 사용하면 여러개의 Step을 사용할 수 있습니다.

@Bean
public Job TestJob(JobRepository jobRepository, PlatformTransactionManager manager){
    return new JobBuilder("TestJob", jobRepository)
            .incrementer(new RunIdIncrementer())
            .start(TestStep(jobRepository, manager))
            .next(TestChunkStep(jobRepository, manager))
            .build();
}

 

 

 

5. Schedule 사용하기

---

SchedulerConfig 생성

@Configuration
@EnableScheduling
@Slf4j
public class SchedulerConfig {
    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private Job job;

    // fixedDelay - 이전 실행이 종료된 시점부터의 간격
    // initialDelay - 스케줄링이 시작되기전 초기 지연 상태
    @Scheduled(fixedDelay = 10000, initialDelay = 2000)
    public void scheduleJob() throws Exception {
        log.info("Job scheduler 시작");
        jobLauncher.run(job, new JobParametersBuilder()
                .addLong("TestJobLauncher", System.nanoTime()).toJobParameters());
        log.info("Job scheduler 끝");
    }
}

- jobLauncher.run ( 실행하려는 Job, JobParameters )

- JobParameters.addLong( "이름", 시간 )

   각 배치 작업에 "이름" + "고유한 타임스탬프 값"을 부여하여, 매번 다른 파라미터를 생성합니다.

 

@Scheduled(cron = "* * * * * *")
cron을 활용해 작업을 예약할 수 있습니다.

첫번째 *부터
"초(0-59)   분(0-59)   시간(0-23)   일(1-31)   월(1-12)   요일(0-6)"
 - 요일 ( 0: 일, 1: 월, 2: 화, 3: 수, 4: 목, 5: 금, 6: 토 )

그외)
? : 설정 값 없음 ( 날짜와 요일에만 사용 가능)
- : 범위를 지정할 때
, : 여러 값을 지정할 때
/ : 증분 값, 즉 초기값과 증가치 설정에 사용
L : 마지막 - 지정할 수 있는 범위의 마지막 값 설정 시 사용 ( 날짜와 요일에서만 사용가능 )
W : 가장 가까운 평일을 설정

zone : cron 표현식을 사용했을 때 사용할 time zone ( 따로 설정 없으면 local의 time zone )
 - @Scheduled(cron = "* * * * * *", zone = "Asia/Seoul")


Example)
@Scheduled(cron = "0 0 18 * * *")                 // 매일 오후 18시에 실행

@Scheduled(cron = "0 0 22 L * *")                 // 매일 마지막날 22시에 실행

@Scheduled(cron = "0 0/5 9-18 * * *")                 // 매일 9시 00분 - 18시 55분 사이에 5분 간격으로 실행

---

BatchConfig 수정

@Bean
public Job TestJob(JobRepository jobRepository, PlatformTransactionManager manager){
    return new JobBuilder("TestJob", jobRepository)
            .incrementer(new RunIdIncrementer())
            .start(TestStep(jobRepository, manager))
            .build();
}

---

결과

- 다음과 같이 반복적으로 실행되는것을 볼 수 있습니다.

 

 

 

마치며

---

최근에 Spring Batch에 관해서 공부하면서 Spring Batch 5.x 버전에 관한 글이 많이 없어서 좋은 코드를 발견하여 이 글을 작성하였습니다.
대용량 처리를 하는데 Spring Batch는 유용한 기능을 많이 제공해주고 있어서 다음에 프로젝트에 적용할 기회가 생기면 사용해볼 생각입니다.

 

https://github.com/helder53/SpringBatch_Example

GitHub - helder53/SpringBatch_Example

 

 

 

---

※ 참고

---

https://europani.github.io/spring/2023/06/26/052-spring-batch-version5.html

[Batch] Spring Batch 5 적용

https://ojt90902.tistory.com/770

Spring Batch : SImpleJob Incrementer 관련

https://dev-coco.tistory.com/176

[Spring Boot] @Scheduled을 이용해 일정 시간 마다 코드 실행하기