CoroutineDispatcher
Coroutine 은 스레드 내에서 실행되며 특정 Coroutine 이 실행중에 IO 요청을 수행하고 응답을 대기 중 일때 코루틴은 중단됩니다. 코루틴이 중단되었을 때 코틀린은 뒷단에서 다른 코루틴을 수행합니다. 그리고 IO 요청에 대한 응답이 와서 처리를 해야 할 때 코루틴은 다시 재개되는데, 이때 재개되는 코루틴은 원래 IO요청을 수행했던 스레드가 아닌 다른 스레드에서 실행될 수도 있습니다.
이렇게 코루틴이 작업을 중단하고 다시 재개할 때 어떤 스레드에서 실행될지를 결정하는 역할을 하는 것은 코루틴 디스패처(CoroutineDispatcher)입니다. 코루틴 디스패처는 코루틴을 어떤 스레드에서 분배할지를 효율적으로 결정하는 역할을 합니다.
즉, 코틀린의 코루틴은 스레드 하나에서만 작업이 쭉 이뤄지지는 않습니다. 그리고 코루틴이 실행될 스레드를 결정하는 것은 코루틴 디스패처(CoroutineDispatcher) 입니다.
Thread 개수의 의미
일반적으로 Thread 하나를 CPU 하나로 비유하는 경우가 있습니다. 이렇게 비유하는 경우는 CPU 중심의 복잡한 연산을 하는 경우에 적합합니다.
일반적인 애플리케이션의 경우 외부 API 요청/응답, Database 접근 등의 연산이 수행됩니다.
이 과정에서 요청을 한 후 응답을 기다리는 놀고 있는 유휴 대기 시간이 많이 생깁니다.
이런 이유로 실무에서 스레드 프로그램을 작성할 때는 IO 작업의 개수가 100개 이더라도 3개 정도의 스레드를 Sheduler 를 이용해서 작업을 스케쥴링해서 운영하는 경우가 많습니다. 즉, 모든 스레드가 1개의 CPU를 온전히 점유하는 것이 아니라 시분할을 통해 CPU 타임을 할당하는 방식입니다.
쿠버네티스에서도 CPU에 CPU Time 을 정의합니다. 예를 들어 CPU 에 대해 1000m 을 지정하면 운영체제에 1초에 CPU 1개 만큼의 시간을 부여받겠다고 요청하는 의미를 가집니다. AI 계산, 수식 계산 등과 같은 작업이 아닌 I/O 작업을 수행하는 서버 애플리케이션의 경우 1초에 CPU 한개를 온전히 소유한다는 것은 굉장히 비싼 축에 속합니다.
이런 이유로 CPU Time 은 일반적으로 250m 과 같이 1초에 0.25 만큼의 CPU 정도를 쓰겠다. 이런 설정을 부여합니다.
이 글을 읽을 때는 Thread 하나로 작업을 수행한다고 해서 온전히 CPU 하나를 의미한다고 생각하지 않으셨으면 합니다.
CoroutineDispatcher 의 종류
코루틴에서 제공하는 대표적인 CoroutineDispatcher 는 Default, IO, Unconfined, Main 이 있습니다.
- Dispatchers.Default
- Dispatchers.IO
- Dispatchers.Unconfined
- Dispatchers.Main
이 중 Dispatchers.Main 은 안드로이드에서만 지원하는 코루틴 디스패처입니다. 따라서 이번 문서에서는 Dispatchers.Main 은 정리하지 않습니다.
Dispatchers.Default
참고 : Dispatchers.Default (opens in a new tab), code : Dispatchers.common.kt (opens in a new tab)
Dispatchers.Default 는 CPU 코어 개수 만큼의 크기의 스레드 풀을 갖는 CoroutineDispatcher 입니다. 만약 CoroutineDispatcher 를 별도로 지정하지 않고 코루틴을 사용하려 한다면 Dispatchers.Default 를 기본으로 사용하게 됩니다.
Dispatchers.Default는 CPU 개수 만큼의 고정된 풀을 사용하기에, 가급적 계산작업의 비중이 높은 CPU 위주의 블로킹연산에 사용하는 것이 좋습니다.
e.g.
package io.chagchagchag.demo.kotlin_coroutine.coroutine_dispatcher
import io.chagchagchag.demo.kotlin_coroutine.helper.logger
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
@OptIn(ExperimentalStdlibApi::class)
fun main(){
val log = logger()
runBlocking {
// (1)
log.info("스레드 == ${Thread.currentThread().name}")
log.info("디스패처 == ${this.coroutineContext[CoroutineDispatcher.Key]}")
// (2)
withContext(Dispatchers.Default){
log.info("스레드 == ${Thread.currentThread().name}")
log.info("디스패처 == ${this.coroutineContext[CoroutineDispatcher.Key]}")
}
}
}
(1)
- main() 내에서 runBlocking 을 했습니다. runBlocking 은 main 스레드의 코루틴 컨텍스트를 그대로 사용하기에 출력결과에서도
스레드 == main라고 나타납니다. - runBlocking 을 호출한 디스패처를 보면
디스패처 == BlockingEventLoop@13a5fe33가 나타납니다. - delay 함수를 사용하지 않고 모든 예제를 main 문과 Dispatchers.Default 기반의 코루틴의 스레드 명을 비교하기 위해
runBlocking{...}을 사용했습니다.
(2)
withContext(Dispatchers.Default){...}을 통해Dispatchers.Default에서 코루틴을 실행하면,스레드 == DefaultDispatcher-worker-1와 같이 출력결과가 나타난 것으로 보아 내부적으로 CoroutineDispatcher 는DefaultDispatcher-worker-1라는 스레드를 할당했음을 알 수 있습니다.- 출력결과로
디스패처 == Dispatchers.Default가 나타났고, runBlocking 을 관리하는 코루틴 디스패처와 withContext 컨텍스트를 실행한 디스패처가 다르다는 것을 확인 가능합니다.
출력결과
18:23:51.714 [main] INFO io...helper.LoggingObject -- 스레드 == main
18:23:51.721 [main] INFO io...helper.LoggingObject -- 디스패처 == BlockingEventLoop@13a5fe33
18:23:51.758 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 스레드 == DefaultDispatcher-worker-1
18:23:51.759 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 디스패처 == Dispatchers.Default
Process finished with exit code 0Dispatchers.IO
참고 : Dispatchers.IO (opens in a new tab), code : Dispatchers.kt (opens in a new tab)
현실세계에서의 서버를 이용한 작업은 대부분 CPU 작업보다는 IO 작업이 많습니다. 외부 API 를 요청해서 결제를 수행한다거나, 데이터를 조회하거나 인증을 수행하는 등 여러가지 요청을 보내고 나서 응답을 받기 전까지의 블로킹이 발생합니다.
따라서 IO 작업 하나에 스레드 하나를 온전히 할당하는 것은 굉장히 큰 낭비가 됩니다. IO 요청이 끝날 때 까지 블로킹되는 대신 응답이 오기 전까지는 다른 작업을 수행하도록 해서 스레드 하나를 여러 작업으로 나눠서 수행하는 것이 중요합니다.
Dispatchers.IO 는 이렇게 blocking I/O 기반의 작업을 수행하기에 적절하도록 코틀린 코루틴에서 제공하는 코루틴 디스패처입니다. 내부적으로는 최대 64개의 스레드까지 가질수 있으며, 스레드 풀은 가변적으로 운영됩니다.
e.g.
package io.chagchagchag.demo.kotlin_coroutine.coroutine_dispatcher
import io.chagchagchag.demo.kotlin_coroutine.helper.logger
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
@OptIn(ExperimentalStdlibApi::class)
fun main(){
val log = logger()
runBlocking {
// (1)
log.info("스레드 == ${Thread.currentThread().name}")
log.info("디스패처 == ${this.coroutineContext[CoroutineDispatcher]}")
// (2)
withContext(Dispatchers.IO){
log.info("스레드 == ${Thread.currentThread().name}")
log.info("디스패처 == ${this.coroutineContext[CoroutineDispatcher]}")
}
}
}(1)
- main() 내에서 runBlocking 을 했습니다. runBlocking 은 main 스레드의 코루틴 컨텍스트를 그대로 사용하기에 출력결과에서도
스레드 == main라고 나타납니다. - runBlocking 을 호출한 디스패처를 보면
디스패처 == BlockingEventLoop@13a5fe33가 나타납니다. - delay 함수를 사용하지 않고 모든 예제를 main 문과 Dispatchers.Default 기반의 코루틴의 스레드 명을 비교하기 위해
runBlocking{...}을 사용했습니다.
(2)
withContext(Dispatchers.Default){...}을 통해Dispatchers.Default에서 코루틴을 실행하면,스레드 == DefaultDispatcher-worker-1와 같이 출력결과가 나타난 것으로 보아 내부적으로 CoroutineDispatcher 는DefaultDispatcher-worker-1라는 스레드를 할당했음을 알 수 있습니다.- 출력결과로
디스패처 == Dispatchers.IO가 나타났고, runBlocking 을 관리하는 코루틴 디스패처와 withContext 컨텍스트를 실행한 디스패처가 다르다는 것을 확인 가능합니다.
출력결과
18:32:05.176 [main] INFO io...helper.LoggingObject -- 스레드 == main
18:32:05.186 [main] INFO io...helper.LoggingObject -- 디스패처 == BlockingEventLoop@13a5fe33
18:32:05.214 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 스레드 == DefaultDispatcher-worker-1
18:32:05.214 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 디스패처 == Dispatchers.IO
Process finished with exit code 0Dispatchers.Unconfined
참고
Unconfined 디스패처는 특정 스레드에 제한되지 않는 코루틴 디스패처입니다. suspend 함수를 재개할 때에 스레드를 사용하는 스레드 정책이 따로 마련되어 있지 않는 방식으로 스레드를 분배합니다. Undefined 디스패처는 중첩 코루틴 실행시에 호출 프레임의 스택오버플로우를 방지하기 위해서 이벤트 루프를 구성합니다.
처음 코루틴을 실행할 때에는 이 코루틴을 실행하기 위해 동작하는 스레드 위에서 호출됩니다. 예를 들면 main() 에서 실행한다면 main 스레드에서 코루틴이 실행됩니다.
그리고 중지된 코루틴을 재개할 때에는 이 중지된 코루틴을 재개를 시작하는 스레드가 이 코루틴의 작업을 재개시킵니다.
쉽게 이야기하면 Dispatchers.Unconfined 는 자신이 실행될 스레드의 종류를 바꿔가면서 무작위로 실행됨을 확인 가능합니다.
Dispatchers.Unconfined 코루틴 디스패처는 스레드가 예측이 불가능해서 일반적인 코드에서 사용하지 않도록 권장하는 편입니다.
스프링 컨트롤러에서는 Dispatchers.Unconfined 기반으로 suspend 함수의 처리를 지원합니다. 아마도 컨트롤러의 요청 처리 방식의 특성 상 호출 프레임에 따라서 호출했던 스레드가 처리를 하기보다는 이벤트 루프 기반으로 수행하는 것이 스택오버플로우 현상을 방지할 수 있기에 이렇게 설정한 것으로 보입니다.
e.g.
package io.chagchagchag.demo.kotlin_coroutine.coroutine_dispatcher
import io.chagchagchag.demo.kotlin_coroutine.helper.logger
import kotlinx.coroutines.*
fun main(){
val log = logger()
runBlocking {
launch(Dispatchers.Unconfined) {
// (1)
log.info("(1) at Unconfined, Thread Name = ${Thread.currentThread().name}")
withContext(Dispatchers.IO){
// (2)
log.info("(2) at Dispatchers.IO, Thread Name = ${Thread.currentThread().name}")
}
// (3)
log.info("(3) at Unconfined, Thread Name = ${Thread.currentThread().name}")
delay(1500)
// (4)
log.info("(4) at Unconfined, ThreadName = ${Thread.currentThread().name}")
}
}
}(1)
- Dispatchers.Unconfined 내에서 스레드 명을 찍어봅니다.
- 결과는
Thread Name = main으로 나타납니다. Unconfined 코루틴 디스패처가 main 스레드를 할당했음을 알 수 있습니다.
(2)
- Dispatchers.IO 내에서 스레드 명을 찍어봅니다.
- 결과는
Thread Name = DefaultDispatcher-worker-1으로 나타납니다. Unconfined 코루틴 디스패처가 실행한 Dispatcher.IO 코루틴 컨텍스트에서는 Dispatchers.IO 코루틴 디스패처가 적용되었음을 유추 가능합니다.
(3)
- Dispatchers.Unconfined 로 다시 돌아와서 스레드 명을 찍어봅니다.
- 결과는
Thread Name = DefaultDispatcher-worker-1으로 나타납니다. 현재 블록이Dispatchers.Unconfined컨텍스트의 영역임에도 이전 컨텍스트가 수행하던 디스패처인Thread Name = DefaultDispatcher-worker-1에서 실행하고 있는 것을 확인 가능합니다.
(4)
- 스레드가 일정시간이 지나면 회수되는지 확인을 위해 1.5초 정도 delay 를 한 후에 Dispatchers.Unconfined 영역에서의 스레드가 다시 재배치 되었는지 확인해봅니다.
- 결과는
ThreadName = kotlinx.coroutines.DefaultExecutor으로 나타납니다. Dispatchers.Unconfined 는 자신이 실행될 스레드의 종류를 바꿔가면서 무작위로 실행됨을 확인 가능합니다.
출력결과
19:16:58.859 [main] INFO io...helper.LoggingObject -- (1) at Unconfined, Thread Name = main
19:16:58.899 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- (2) at Dispatchers.IO, Thread Name = DefaultDispatcher-worker-1
19:16:58.899 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- (3) at Unconfined, Thread Name = DefaultDispatcher-worker-1
19:17:00.426 [kotlinx.coroutines.DefaultExecutor] INFO io...helper.LoggingObject -- (4) at Unconfined, ThreadName = kotlinx.coroutines.DefaultExecutor
Process finished with exit code 0Dispatchers.Main
Dispatchers.Main 은 안드로이드에서만 지원하는 코루틴 디스패처입니다. 따라서 이번 문서에서는 Dispatchers.Main 은 정리하지 않습니다.
Dispatchers.IO 와 Dispatchers.Default
Dispatchers.IO와 Dispatchers.Default 는 스레드 풀을 공유합니다.
e.g. 1 :
package io.chagchagchag.demo.kotlin_coroutine.coroutine_dispatcher
import io.chagchagchag.demo.kotlin_coroutine.helper.logger
import kotlinx.coroutines.*
@OptIn(ExperimentalStdlibApi::class)
fun main(){
val log = logger()
runBlocking {
// (1)
log.info("스레드 == ${Thread.currentThread().name}")
log.info("디스패처 == ${this.coroutineContext[CoroutineDispatcher.Key]}")
// (2)
withContext(Dispatchers.Default){
log.info("스레드 == ${Thread.currentThread().name}")
log.info("디스패처 == ${this.coroutineContext[CoroutineDispatcher.Key]}")
}
// (3)
withContext(Dispatchers.IO){
log.info("스레드 == ${Thread.currentThread().name}")
log.info("디스패처 == ${this.coroutineContext[CoroutineDispatcher.Key]}")
}
// (4)
CoroutineScope(CoroutineName("배고파요")).launch {
log.info("스레드 == ${Thread.currentThread().name}")
log.info("디스패처 == ${this.coroutineContext[CoroutineDispatcher.Key]}")
}
}
}(1)
- main 스레드 안에서 스레드 명, CoroutineDispatcher 를 출력해봅니다.
- 출력결과는 스레드는
main으로, 디스패처는BlockingEventLoop@13a5fe33이 출력됩니다. runBlocking 내의 코루틴 디스패처를 접근했기 때문에BlockingEventLoop@13a5fe33이 조회되었습니다.
(2)
- Dispatchers.Default 코루틴 컨텍스트 안에서 스레드명, CoroutineDispatcher 를 출력해봅니다.
- 출력결과는 스레드는
DefaultDispatcher-worker-1으로, 디스패처는Dispatchers.Default으로 나타났습니다.
(3)
- Dispatchers.IO 코루틴 컨텍스트 안에서 스레드명, CoroutineDispatcher 를 출력해봅니다.
- 출력결과는 스레드는
DefaultDispatcher-worker-1으로, 디스패처는Dispatchers.IO로 나타났습니다.
(4)
- 코루틴 디스패처를 별도로 명시하지 않은 코루틴 컨텍스트 안에서 스레드명, CoroutineDispatcher 를 출력해봅니다.
- 출력결과는 스레드는
DefaultDispatcher-worker-1으로, 디스패처는Dispatchers.Default으로 나타났습니다.
출력결과
19:32:50.451 [main] INFO io...helper.LoggingObject -- 스레드 == main
19:32:50.458 [main] INFO io...helper.LoggingObject -- 디스패처 == BlockingEventLoop@13a5fe33
19:32:50.498 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 스레드 == DefaultDispatcher-worker-1
19:32:50.499 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 디스패처 == Dispatchers.Default
19:32:50.507 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 스레드 == DefaultDispatcher-worker-1
19:32:50.508 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 디스패처 == Dispatchers.IO
19:32:50.522 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 스레드 == DefaultDispatcher-worker-1
19:32:50.523 [DefaultDispatcher-worker-1] INFO io...helper.LoggingObject -- 디스패처 == Dispatchers.Default
Process finished with exit code 0e.g. 2 : Dispatchers.IO, Dispatchers.Default 의 스레드 공유 예제
이번에는 아래 예제 처럼 비교적 큰 수인 2000 개의 작업을 launch 를 통해서 코루틴을 수행하면 worker 스레드가 64개 이상으로 늘어나고, 출력결과를 보면 모두 DefaultDispatcher-worker-x 와 같은 포맷으로 출력됨을 확인할 수 있습니다.
package io.chagchagchag.demo.kotlin_coroutine.coroutine_dispatcher
import io.chagchagchag.demo.kotlin_coroutine.helper.logger
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
fun main (){
val log = logger()
runBlocking {
for(i in 1 until 2000){
launch (Dispatchers.Default){
log.info("현재 스레드 : ${Thread.currentThread().name}")
}
launch (Dispatchers.IO){
log.info("현재 스레드 : ${Thread.currentThread().name}")
}
}
}
}ExecutorCoroutineDispatcher
Dispatchers.Default 는 너무 CPU Bound 작업에 특화되어 있고, Dispatchers.IO 는 너무 스레드 자원을 크게 사용한다는 느낌이 들 때가 있습니다. 스레드 풀을 조금 더 경량화 해서 잡을 수도 있을 것 같고, 더 효율적으로 스레드를 사용하도록 스케쥴링 기반으로 전환할 수도 있습니다.
이런 경우 ExecutorCoroutineDispatcher 를 사용해서 원하는 커스텀 설정이 적용된 CoroutineDispatcher를 만들어낼 수 있습니다.
ExecutorCoroutineDispatcher 를 사용하면 특정 크기의 스레드 풀을 갖는 Dispatcher 를 생성 가능하고 직접 설정한 Executor를 주입해서 사용 가능합니다.
만약 저라면 ExecutorCoroutineDispatcher 를 용도별로 분류해서 적극적으로 활용하지 않을까 싶습니다.
newFixedThreadPoolContext
e.g.
package io.chagchagchag.demo.kotlin_coroutine.coroutine_dispatcher
import io.chagchagchag.demo.kotlin_coroutine.helper.logger
import kotlinx.coroutines.newFixedThreadPoolContext
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
fun main(){
val log = logger()
runBlocking {
log.info("START")
val fixedDispatcher = newFixedThreadPoolContext(1, "messageSender")
withContext(fixedDispatcher){
log.info("스레드 : ${Thread.currentThread().name}, 디스패처 : $fixedDispatcher")
fixedDispatcher.close()
log.info("END")
}
}
}출력결과
14:42:55.186 [main] INFO io...helper.LoggingObject -- START
14:42:55.240 [messageSender] INFO io...helper.LoggingObject -- 스레드 : messageSender, 디스패처 : java.util.concurrent.ScheduledThreadPoolExecutor@6ddf90b0[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]
14:42:55.241 [messageSender] INFO io...helper.LoggingObject -- END
Process finished with exit code 0newSingleThreadContext
e.g.
package io.chagchagchag.demo.kotlin_coroutine.coroutine_dispatcher
import io.chagchagchag.demo.kotlin_coroutine.helper.logger
import kotlinx.coroutines.newSingleThreadContext
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
fun main() {
val log = logger()
runBlocking {
log.info("START")
val singleDispatcher = newSingleThreadContext("sumCartDispatcher")
withContext(singleDispatcher) {
log.info("스레드 : ${Thread.currentThread().name}, 디스패처 : $singleDispatcher")
singleDispatcher.close()
log.info("END")
}
}
}
출력결과
14:45:04.233 [main] INFO io...helper.LoggingObject -- START
14:45:04.282 [sumCartDispatcher] INFO io...helper.LoggingObject -- 스레드 : sumCartDispatcher, 디스패처 : java.util.concurrent.ScheduledThreadPoolExecutor@5a8e6209[Running, pool size = 1, active threads = 1, queued tasks = 0, completed tasks = 0]
14:45:04.283 [sumCartDispatcher] INFO io...helper.LoggingObject -- END
Process finished with exit code 0e.g. fixed, single
package io.chagchagchag.demo.kotlin_coroutine.coroutine_dispatcher
import io.chagchagchag.demo.kotlin_coroutine.helper.logger
import kotlinx.coroutines.*
fun main(){
val log = logger()
runBlocking {
val fixed = newFixedThreadPoolContext(2,"messageSenderDispatcher")
val singleThreadDispatcher = newSingleThreadContext("sumCartDispatcher")
val job = launch (singleThreadDispatcher){
log.info("스레드 1 : ${Thread.currentThread().name}")
withContext(fixed){
log.info("스레드 2 : ${Thread.currentThread().name}")
withContext(Dispatchers.IO){
log.info("스레드 3 : ${Thread.currentThread().name}")
withContext(singleThreadDispatcher){
log.info("스레드 4 : ${Thread.currentThread().name}")
}
}
}
}
job.join()
fixed.close()
singleThreadDispatcher.close()
}
}확인해보면 모두 독자적으로 제공받은 Dispatcher 를 그대로 사용하고 있음을 확인 가능합니다.
출력결과
14:34:37.155 [sumCartDispatcher] INFO io...helper.LoggingObject -- 스레드 1 : sumCartDispatcher
14:34:37.163 [messageSenderDispatcher-1] INFO io...helper.LoggingObject -- 스레드 2 : messageSenderDispatcher-1
14:34:37.176 [DefaultDispatcher-worker-2] INFO io...helper.LoggingObject -- 스레드 3 : DefaultDispatcher-worker-2
14:34:37.178 [sumCartDispatcher] INFO io...helper.LoggingObject -- 스레드 4 : sumCartDispatcher
Process finished with exit code 0