단일 스레드 방식의 한계점
외부 서버 API로부터 데이터를 받아와 DB에 적재를 해야한다고 하자
API 서버에 데이터를 요청할 때 모든 데이터가 아닌 page 단위로 요청해야하는 상황이라면 단건씩 요청을 해야한다.
만약 페이지 수가 1000개라면, 단일 스레드에서 1000개 페이지에 데이터를 요청하고, 데이터베이스에 적재해야한다.
위 작업을 단일 스레드에서 실행하면 모든 작업을 순차적으로 처리하기 때문에 페이지가 많을 수록 대기시간이 오래걸린다.
이를 해결해 줄 수 있는 방법은 Partitioner를 사용하여 멀티 스레드로 처리하는 것이다.
구조
파티셔너는 Step이 각 스레드에 배정되어 실행될 수 있도록 분배하는 역할이다.
구체적인 분배 방법은 Partitioner 인터페이스를 구현하여 정의할 수 있다.
방법
파티셔너 생성
class MemberPartitioner : Partitioner {
override fun partition(gridSize: Int): MutableMap<String, ExecutionContext> {
return (1..gridSize).associate {
val context = ExecutionContext()
context.put("maxPage", it * 100)
"partition$it" to context
}.toMutableMap()
}
}
- gridSize: 파티션을 얼마나 나눌지 참고하는 값 (꼭 사용하지 않아도 된다.)
- 위 코드는 스텝이 처리할 수 있는 임의의 값(maxPage)을 context에 담고 있다.
- 만약 총 페이지가 5,000 페이지라고 가정하고 gridSize 가 10이라면,
스텝에 전달될 각 컨텍스트에는 maxPage 값이 500, 1000, 1500, …, 5000 담겨지도록 만들어 볼 수 있다.
- 만약 총 페이지가 5,000 페이지라고 가정하고 gridSize 가 10이라면,
파티셔너 등록
@Bean
fun partitionerStep(): Step {
return StepBuilder("step", jobRepository)
.partitioner("partitioner", MemberPartitioner()) // MemberPartitioner 파티셔너 지정
.gridSize(5) // 파티셔너가 참고할 사이즈를 정의
.step(step1()) // 분배된 파티션 context를 가지는 스텝(step1)을 정의
.taskExecutor(taskExecutor())
.build()
}
@Bean
fun taskExecutor(): ThreadPoolTaskExecutor { // 스텝이 실행될 스레드 풀 정의
val taskExecutor = ThreadPoolTaskExecutor()
taskExecutor.corePoolSize = 10
taskExecutor.maxPoolSize = 10
taskExecutor.initialize()
return taskExecutor
}파티셔닝된 스텝 등록
@Bean
fun step1(): Step {
return StepBuilder("step1", jobRepository)
.tasklet(memberTask(null), transactionManager)
.build()
}
@Bean
@StepScope
fun memberTask(@Value("#{stepExecutionContext['maxPage']}") max: Int?): Tasklet {
return MemberTasklet(max!!)
}
class MemberTasklet(
private val max: Int
) : Tasklet {
override fun execute(contribution: StepContribution, chunkContext: ChunkContext): RepeatStatus? {
// val max = contribution.stepExecution.executionContext["maxPage"] as Int
println("Hello - ${Thread.currentThread().name}")
println("$max - ${Thread.currentThread().name}")
return RepeatStatus.FINISHED
}
}파티셔너로부터 전달받은 값을 출력하는 간단한 tasklet 스텝이다.
실제로는 maxPage 수의 API 요청을 하고 데이터베이스에 적재하는 스텝이 될 수 있다.
- StepScope: Tasklet 스텝에서 StepContribution을 통해 컨텍스트를 가져올 수 있고, @StepScope를 사용하여 쉽게 컨텍스트 값을 가져올 수 있다. 두 가지 방법 중 선호하는 방식을 택하면 된다.
- 스텝이 어떤 스레드에 배정되어 실행되는지 확인할 수 있게 스레드 이름을 출력
결과
2024-04-01T13:15:34.161+09:00 INFO 79068 --- [batch-multithread] [ main] o.s.b.c.l.support.SimpleJobLauncher : Job: [SimpleJob: [name=firstJob]] launched with the following parameters: [{}]
2024-04-01T13:15:34.184+09:00 INFO 79068 --- [batch-multithread] [ main] o.s.batch.core.job.SimpleStepHandler : Executing step: [step]
Hello - taskExecutor-1
Hello - taskExecutor-10
1000 - taskExecutor-1
Hello - taskExecutor-9
200 - taskExecutor-10
300 - taskExecutor-9
Hello - taskExecutor-8
500 - taskExecutor-8
Hello - taskExecutor-7
800 - taskExecutor-7
Hello - taskExecutor-6
400 - taskExecutor-6
Hello - taskExecutor-2
600 - taskExecutor-2
Hello - taskExecutor-3
900 - taskExecutor-3
Hello - taskExecutor-4
100 - taskExecutor-4
Hello - taskExecutor-5
700 - taskExecutor-5
2024-04-01T13:15:34.226+09:00 INFO 79068 --- [batch-multithread] [taskExecutor-10] o.s.batch.core.step.AbstractStep : Step: [step1:partition2] executed in 25ms
2024-04-01T13:15:34.226+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-1] o.s.batch.core.step.AbstractStep : Step: [step1:partition10] executed in 27ms
2024-04-01T13:15:34.227+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-9] o.s.batch.core.step.AbstractStep : Step: [step1:partition3] executed in 26ms
2024-04-01T13:15:34.227+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-5] o.s.batch.core.step.AbstractStep : Step: [step1:partition7] executed in 27ms
2024-04-01T13:15:34.227+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-8] o.s.batch.core.step.AbstractStep : Step: [step1:partition5] executed in 26ms
2024-04-01T13:15:34.228+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-7] o.s.batch.core.step.AbstractStep : Step: [step1:partition8] executed in 27ms
2024-04-01T13:15:34.228+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-6] o.s.batch.core.step.AbstractStep : Step: [step1:partition4] executed in 27ms
2024-04-01T13:15:34.228+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-2] o.s.batch.core.step.AbstractStep : Step: [step1:partition6] executed in 28ms
2024-04-01T13:15:34.228+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-3] o.s.batch.core.step.AbstractStep : Step: [step1:partition9] executed in 28ms
2024-04-01T13:15:34.229+09:00 INFO 79068 --- [batch-multithread] [ taskExecutor-4] o.s.batch.core.step.AbstractStep : Step: [step1:partition1] executed in 28ms
2024-04-01T13:15:34.231+09:00 INFO 79068 --- [batch-multithread] [ main] o.s.batch.core.step.AbstractStep : Step: [step] executed in 46ms
2024-04-01T13:15:34.235+09:00 INFO 79068 --- [batch-multithread] [ main] o.s.b.c.l.support.SimpleJobLauncher : Job: [SimpleJob: [name=firstJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 63ms
각 스레드별 파티셔닝된 데이터가 출력되는 것을 확인할 수 있다.
