스프링 부트 배치 - 심화

1. 스프링 배치 심화학습
2. 멀티 스레드로 여러 개의 Step 실행하기

---

1. 스프링 배치 심화학습

1. 다양한 ItemReader 구현 클래스
2. 다양한 ItemWriter 구현 클래스
3. JobParameter 사용하기
4. 테스트 시에만 H2 DB를 사용하도록 설정하기
5. 청크 지향 프로세싱
6. 배치의 인터셉터 Listener 설정하기
7. 어노테이션 기반으로 Listener 설정하기
8.  Step의 흐름을 제어하는 Flow

1.1 다양한 ItemReader 구현 클래스

리스트 타입으로 Reader를 구현한 ListItemReader 객체가 있습니다.

전에 기본편에서 만들었던 QueueItemReader 객체와 동일한 역할을 수행합니다.

ListItemReader 객체를 사용하면 모든 데이터를 한번에 가져와 메모리에 올려놓고 read() 메서드로 하나씩 배치 처리 작업을 수행할 수 있습니다.

그런데 수백, 수천을 넘어 수십만 개 이상의 데이터를 한번에 가져와 메모리에 올려놓아야 할 때는 어떻게 해야 할까요?

이때는 배치 프로젝트에서 제공하는 PagingItemReader 구현체를 사용할 수 있습니다.

구현체는 크게 JdbcPagingItemReader, JpaPagingItemReader, HibernatePagingItemReader가 있습니다. 

JpaPagingItemReader를 사용해보겠습니다.

JpaPagingItemReader에는 지정한 데이터 크기만큼 DB에서 읽어오는 setPageSize() 메서드 기능이 있습니다.

데이터를 지정한 단위로 가져와 배치 처리를 수행할 수 있습니다.

1. @Bean(destroyMethod="") -> 스프링에서 destroyMethod를 사용해 삭제할 빈을 자동으로 추적합니다.

destoryMethod=""와 같이 하여 기능을 사용하지 않도록 설정하면 실행 시 출력되는 warning 메시지를 삭제할 수 있습니다.

2. jpaPagingItemReader.setQueryString() -> jpaPagingItemReader를 사용하려면 쿼리를 직접 짜서 실행하는 방법밖에 없습니다. 

3. jpaPagingItemReader.setParameterValues(map) -> updatedDate, status 파라미터를 Map에 추가해 사용할 파라미터를 설정합니다.

 

JpaPagingItemReader 주의사항

inacticeJobStep()에서 설정한 청크 단위(커밋 단위)가 5라고 가정하면 Item5개를 writer까지 배치 처리를 진행하고 저장한다고 해봅시다. 저장된 데이터를 바탕으로 다음에 다시 지정한 크기로 새 인덱스를 할당해 읽어 와야 하는데 이전에 진행한 5라는 인덱스값을 그대로 사용해 데이터를 불러오도록 로직이 짜여 있어서 문제가 됩니다.

예를 들어 청크 단위로 Item 5개를 커밋하고 다음 청크 단위로 넘어가야 하는 경우를 가정하겠습니다.

하지만 entityManager에서 앞서 처리된 Item 5개 때문에 새로 불러올 Item의 인덱스 시작점이 5로 설정되어 있게 됩니다. 그러면 쿼리 요청 시 offset 5(인덱스값), limit 5(지정한 크기 단위)이므로 개념상 바로 다음 청크 단위(Item 5개)인 Item을 건너뛰는 상황이 발생합니다.

이러한 상황에서 가장 간단한 해결 방법은 조회용 인덱스값을 항상 0으로 반환하는 겁니다.

0으로 반환하면 Item 5개를 수정하고 다음 5개를 건너뛰지 않고 원하는 순서/청크 단위로 처리가 가능해집니다.

 

1.2 다양한 ItemWriter 구현 클래스

JpaItemWriter는 별도로 저장 설정을 할 필요 없이 제네릭에서 저장할 타입을 명시하고 EntityManagerFactory만 설정하면 Processor에서 넘어온 데이터를 청크 단위로 저장합니다.

1.3 JobParameter 사용하기

테스트 코드에 JobParameter를 생성해 JobLauncher에 전달하게끔 수정합니다.

1. new Date(); -> Date 타입은 JobParameter에서 허용하는 파라미터 중 하나입니다.

2. jobLauncherTestUtils.launchJob(new JobParametersBuilder().addDate("nowDate", nowDate).toJobParameters()) ->

JobParametersBuilder를 사용하면 간편하게 JobParameters를 생성할 수 있습니다. JobParameters는 여러 JobParameter를 받는 객체입니다. JobLauncher를 사용하려면 JobParameters가 필요합니다.

 

1.4 테스트 시에만 H2 DB를 사용하도록 설정하기

@AutoConfigureTestDatabase(connection = EmbeddedDatabaseConnection.H2) 어노테이션을 사용하면 테스트 시에는 H2를 할당하게 처리할 수 있습니다.

1.5 청크 지향 프로세싱

청크 지향 프로세싱은 트랜잭션 경계 내에서 청크 단위로 데이터를 읽고 생성하는 프로그래밍 기법입니다. 

청크란 아이템이 트랜잭션에서 커밋되는 수를 말합니다.

read한 데이터 수가 지정한 청크 단위와 일치하면 wirte를 수행하고 트랜잭션을 커밋합니다.

청크 지향 프로세싱의 이점은 1000여 개의 데이터에 대해 배치 로직을 실행한다고 가정합시다. 청크로 나누지 않았을 때는 하나만 실패해도 다른 성공한 999개의 데이터가 롤백됩니다. 그런데 청크 단위를 10으로 해서 배치 처리를 하면 도중에 배치 처리에 실패하더라도 다른 청크는 영향을 받지 않습니다.

 

1.6 배치의 인터셉터 Listener 설정하기

배치 흐름에서 전후 처리를 하는 Listener를 설정할 수 있습니다.

구체적으로 Job의 전후 처리, Step의 전후 처리, 각 청크 단위에서의 전후 처리 등 세세한 과정 실행 시 특정 로직을 할당해 제어할 수 있습니다.

스프링 배치에서는 Job의 Listener로 JobExecutionListener 인터페이스를 제공합니다.

1. @Slf4j -> 필드에 로그 객체를 따로 생성할 필요 없이 로그 객체를 사용할 수 있도록 설정하는 롬복 어노테이션 

 

- Job 설정에 Listener 등록하기

InactiveUserJobConfig.java

1.7 어노테이션 기반으로 Listener 설정하기

InacticeStepListener.java

- Step 설정에 Listener 등록하기

 

1.8 Step의 흐름을 제어하는 Flow

스프링 배치는 세부적인 조건에 대한 흐름을 제어하는 Flow를 제공합니다.

흐름에서 조건에 해당하는 부분을 JobExecutionDecider 인터페이스를 사용해 구현할 수 있습니다.

JobExecutionDecider 인터페이스는 decide() 메서드 하나만 제공합니다.

1. start(new InactiveJobExecutionDecider()) -> start()로 설정해 맨 처음 시작하도록 지정합니다.

2. on(FlowExecutionStatus.FAILED.getName()).end() -> failed가 반환되면 end()를 사용해 곧바로 끝나도록 설정합니다.

3. on(FlowExecutionStatus.COMPLETED.getName()).to(inactiveJobStep) -> completed가 반환되면 기존에 설정한 inactiveJobStep을 실행하도록 설정합니다.

4. start(inactiveJobFlow) -> inactiveUserJob 시작 시 Flow를 거쳐 Step을 실행하도록 inactiveJobFlow를 start()에 설정합니다.

 

2. 멀티 스레드로 여러 개의 Step 실행하기

1. TaskExecutor를 사용해 여러 Step 동작시키기

 

2.1 TaskExecutor를 사용해 여러 Step 동작시키기

TaskExcutor 인터페이스는 멀티 스레드로 Step을 실행하는 가장 기본적인 방법입니다.

Task는 Runnable 인터페이스를 구현해 각각의 스레드가 독립적으로 실행되도록 작업을 할당하는 객체입니다.