(2) 스프링 비동기 처리를 위한 ThreadPoolTaskExecutor와 테스트

스프링에서 @EnableAsync 어노테이션과 비동기로 동작하기 바라는 메서드에 @Async 어노테이션을 사용해주면, 그 함수는 비동기로 작동할 수 있다.

이전 글에서 비동기 처리를 위한 TaskExecutor를 설정해주지 않으면 SimpleAsyncTaskExecutor가 사용된다고 했는데, 과연 그럴까 ? 그리고 많은 요청이 들어왔을 때 스레드 풀이 어떻게 동작하는지 테스트해보자.

---

ThreadPoolTaskExecutor 옵션

• Core Pool Size : 항상 유지할 최소 스레드의 개수
• Max Pool Size : 스레드 풀이 가질 수 있는 최대 스레드의 개수, (Core Pool Size + 큐의 용량)보다 많은 요청이 들어오면 새로운 스레드를 생성하고 Max Pool Size보다 큰 개수의 스레드를 가질 수 없다.
• Queue Capacity : 큐에서 대기할 요청의 최대 개수
• Keep Alive Seconds : 큐의 용량보다 많은 요청에 의해 새롭게 생성된 스레드가 대기 상태로 머무를 수 있는 시간이며, 이 시간이 지나면 스레드는 종료된다.

ThreadPoolTaskExecutor의 동작

 

1. 스레드 풀의 core pool size 이내의 스레드가 작업을 처리하고 있으면, 해당 스레드에 작업을 할당하여 처리한다.
2. core pool size 만큼의 스레드가 작업을 처리하고 있고 큐의 용량이 가득차지 않았으면, 큐에 작업을 대기시킨다.
3. 큐의 용량이 가득 찼고, 현재 스레드의 개수가 max pool size 이하인 경우 새로운 스레드를 생성하여 해당 작업을 처리한다.
4. 큐의 용량이 가득 찼고, 현재 스레드의 개수가 max pool size인 경우 RejectedExecutionException 예외가 발생한다.

---

Controller와 Service

Controller에서 요청을 받으면, Service의 비동기 함수를 반복문만큼 실행해본다.

@RestController
@RequiredArgsConstructor
public class AsyncController {

    private final AsyncService asyncService;

    @GetMapping("/async")
    public String asyncController() throws InterruptedException {

        for(int i = 1 ; i <= 1000; i++) {
            asyncService.async_func(i);
        }
        return "hello";
    }
}

@Service
@Slf4j
@RequiredArgsConstructor
public class AsyncService {

    @Async
    public void async_func(int i) throws InterruptedException {
        log.info("i: " + i);
        Thread.sleep(10000);
    }

}

---

ThreadPoolTaskExecutor를 설정하지 않은 경우

실행 결과

1000개의 스레드를 생성하고, 리소스의 부족으로 예외가 발생할 것으로 예상했다.

하지만 실제 결과는 8개의 스레드만을 사용해서 8개의 작업으로 나눠서 처리했다. 또한 SimpleAsyncTaskExecutor를 사용하지 않고, core pool size가 8인 ThreadPoolTaskExecutor을 사용했다.

 

default ThreadPoolTaskExecutor

디버깅 해보면 corePoolSize가 8인 ThreadPoolTaskExecutor을 사용한 걸 알 수 있다.

 

@SpringBootApplication으로 스프링이 실행될 때 수많은 빈이 등록 되는데, 그 중 하나가 TaskExecutorConfigurations이다.

TaskExecutionProperties

 

ThreadPoolTaskExecutor의 속성들은 TaskExecutionProperties에서 설정되어 있고, 여기서 coreSize = 8, threadNamePrefix = "task-", maxPoolSize = Integer.MAX_VALUE, queueCapacity = Integer.MAX_VALUE로 설정된 것을 확인할 수 있다.

그렇기 때문에 1000개의 요청이 들어와도 모두 큐에 넣어 대기시키기 때문에 오류가 터지지 않고, 8개의 스레드로 나눠서 처리하는 것이다. 또한 스프링의 경우 SimpleAsyncTaskExecutor가 사용되지만, 스프링 부트의 경우 auto configure에 의해 ThreadPoolTaskExecutor가 Bean으로 주입되어 default로 사용된다.

 

---

ThreadPoolTaskExecutor를 직접 등록

@EnableAsync
@Configuration
public class AsyncConfig implements AsyncConfigurer {
    @Override
    public Executor getAsyncExecutor() {

        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

        executor.setCorePoolSize(2);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(10);
        executor.setKeepAliveSeconds(30);
        executor.setThreadNamePrefix("thread-pool");

        executor.initialize();

        return executor;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return AsyncConfigurer.super.getAsyncUncaughtExceptionHandler();
    }
}

 

실험 1

• Core Pool Size  = 2
• Max Pool Size = 10
• Queue Capacity = 10
• 20개의 요청
• 예상 결과 : 1, 2번의 요청, 3~12번의 요청은 큐에 대기, 13~20번 요청은 새로운 스레드를 만들어 처리

실행 결과

예상과 같게 1, 2, 3~20번 요청이 거의 동시에 실행되었고, 10초가 지나서 요청이 끝난 후 큐에서 대기하고 있던 3~12번 요청이 처리되었다

 

실험 2

• Core Pool Size  = 2
• Max Pool Size = 10
• Queue Capacity = Integer.MAX_VALUE
• 1000개의 요청
• 예상 결과 : 큐의 용량이 Integer.MAX_VALUE이기 때문에 새로운 스레드를 생성하지 않고, 2개씩 모든 요청을 처리한다.

실행 결과

예상 결과대로 새로운 스레드를 생성하지 않고, 큐에 모두 대기시켜 2개씩 순차적으로 처리했다.

 

실험 3

• Core Pool Size  = 2
• Max Pool Size = 10
• Queue Capacity =10
• 21개의 요청
• 예상 결과 : 1, 2번 요청 실행, 3~12번 요청 큐에서 대기, 13~20번 요청 새로운 스레드 생성하여 실행, 21번 요청에서 TaskRejectedException 예외 발생

 

예상 결과대로 1, 2, 13~20번 요청이 먼저 실행되고, 3~12번 요청은 큐에 대기하며,

21번 요청에서는 큐의 용량이 꽉 찼고 스레드 생성 개수 또한 최대이기 때문에 예외가 발생했다.

 

---

결론

스프링 부트에서는 auto configure에 의해 ThreadPoolTaskExecutor가 Bean으로 동록되어, 과도한 스레드의 생성으로 인한 자원 부족을 예방한다.

또한 서버의 트래픽과 스펙 등을 따져 ThreadPoolTaskExecutor를 설정해야 할 것이다.