spring 에서 비동기 처리를 위한 TreadPoolTaskExecutor 알아보기

Spring ThreadPoolTaskExcutor 는 java.util.concurrent.ThreadPoolExecutor 의 추상화 한것을
Spring 에서 제공하는 JavaBean 입니다.

 

• corePoolSize
• maxPoolSize
• queueCapacity
• allowCoreThreadTimeOut
• keepAliveSeconds

와 같은 속성을 통해 커스텀하게 사용 할 수 있습니다.

 

corePoolSize vs maxPoolSize

 

corePoolSize

corePoolSize 는 timeout 없이 active status 를 유지하기 위한 최소한의 worker 수입니다.

ThreadPoolTaskExecutor 는 구성된 설정을 java.util.concurrent.ThreadPoolExecutor 에 위임합니다.

정확히 말하자면 모든 쓰레드가 timeout 이 될 수 있습니다.

 allowCoreThreadTimeOut  을 true 로 설정한 경우 corePoolSize 값을 효과적으로 0으로 설정 합니다 .

 

maxPoolSize

corePoolSize 와 달리 maxPoolSize 는 생성될 수 있는 최대 쓰레드 개수를 정의합니다. 

corePoolSize 와 동일하게 구성된 설정을  java.util.concurrent.ThreadPoolExecutor 에 위임합니다.

maxPoolSize 는 ThreadPoolTaskExecutor의 대기열의 항목 수가 queueCapacity 를 초과하는 경우에만 새 스레드를 생성 한다는 점 에서 queueCapacity 에 의존 합니다.

 

Thread 를 만드는 시점

ThreadPoolTaskExecutor에 새로운 task 를 submit 할때 pool 에 사용가능한 쓰레드가 있더라도 corePoolSize 보다 적은 쓰레드가 있거나 maxPoolSize 보다 적은 쓰레드가 실행중이고 queueCapacity 에 의해 정의된 큐사이즈가 가득차면 새 쓰레드를 생성합니다.

 

예시

 

먼저 ThreadPoolTaskExecutor 에서 startThreads 라는 새 스레드를 실행하는 메서드가 있다고 가정해 보겠습니다 .

package com.example.threadexample

import java.util.concurrent.CountDownLatch
import java.util.concurrent.ThreadLocalRandom
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor

@SpringBootApplication
class ThreadExampleApplication

fun main(args: Array<String>) {
    runApplication<ThreadExampleApplication>(*args)
}

fun startThreads(taskExecutor: ThreadPoolTaskExecutor, countDownLatch: CountDownLatch, numThreads: Int) {
    for (i in 1 .. numThreads) {
        taskExecutor.execute {
            try {
                Thread.sleep(100L * ThreadLocalRandom.current().nextLong(1, 10))
                countDownLatch.countDown()
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()

            }
        }
    }
}

 

하나의 쓰레드의 corePoolsize , maxPoolSize 및 queueCapacity 를 제한없이 설정하고  ThreadPoolTaskExecutor 의 기본 구성을 테스트해 보겠습니다 . 

결과적으로 아무리 많은 작업을 하더라도 하나의 쓰레드만 실행되었습니다.

package com.example.threadexample

import java.util.concurrent.CountDownLatch
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor

class ThreadExampleApplicationTests {

    @Test
    fun whenCorePoolSizeFiveAndMaxPoolSizeTneThenFiveThread() {
        val threadPoolTaskExecutor = ThreadPoolTaskExecutor()
        threadPoolTaskExecutor.afterPropertiesSet()

        val countDownLatch = CountDownLatch(10)
        startThreads(threadPoolTaskExecutor, countDownLatch, 10)

        while (countDownLatch.count > 0) {
            if (threadPoolTaskExecutor.poolSize == 10) {
                assertEquals(1, threadPoolTaskExecutor.poolSize)
            }
        }
    }

}

 

이제 corePoolSize 를 최대 5개의 쓰레드로 변경하고 정상적으로 동작하는지 확인합니다.

@Test
fun whenCorePoolSizeFiveAndMaxPoolSizeTneThenFiveThread() {
    val threadPoolTaskExecutor = ThreadPoolTaskExecutor()
    threadPoolTaskExecutor.corePoolSize = 5
    threadPoolTaskExecutor.afterPropertiesSet()

    val countDownLatch = CountDownLatch(10)
    startThreads(threadPoolTaskExecutor, countDownLatch, 10)

    while (countDownLatch.count > 0) {
        if (threadPoolTaskExecutor.poolSize == 10) {
            assertEquals(1, threadPoolTaskExecutor.poolSize)
        }
    }
}

마찬가지로 corePoolSize 를 5로 유지하면서 maxPoolSize 를 10으로 늘릴수 있습니다. 결과적으로 5개의 쓰레드만 실행됩니다.

queueCapacity 의 용량이 limit 이 없기때문에 5개의 쓰레드만 실행됩니다.

 

 

마지막으로 queueCapacity 를 10으로 증가시키고 20개의 쓰레드를 시작합니다.

@Test
fun whenCorePoolSizeFiveAndMaxPoolSizeTneThenFiveThread() {
    val threadPoolTaskExecutor = ThreadPoolTaskExecutor()
    threadPoolTaskExecutor.corePoolSize = 5
    threadPoolTaskExecutor.maxPoolSize = 10
    threadPoolTaskExecutor.queueCapacity = 10
    threadPoolTaskExecutor.afterPropertiesSet()

    val countDownLatch = CountDownLatch(20)
    startThreads(threadPoolTaskExecutor, countDownLatch, 20)

    while (countDownLatch.count > 0) {
        assertEquals(10, threadPoolTaskExecutor.poolSize)
    }
}

총 10개의 쓰레드가 실행됩니다.

 

참고자료

https://www.baeldung.com/java-threadpooltaskexecutor-core-vs-max-poolsize

ThreadPoolTaskExecutor corePoolSize vs. maxPoolSize | Baeldung