<Coroutine> 2. CoroutineDispatcher와 CoroutineBuilder 그리고 async

참고 & 출처 : https://www.yes24.com/Product/Goods/125014350

코틀린 코루틴의 정석 - 예스24

가.  코루틴 빌더

 a.  무엇인가?

  -  코루틴을 생성하는데 사용하는 함수를 코루틴 빌더라고 부른다. ex) runBlocking, launch

  -  코루틴 빌더는 코루틴을 만들고 코루틴을 추상화한 Job 객체를 생성한다. 

 

 b.  코루틴 Job 

  -  Job 인터페이스 내에는 상태를 확인 할 수 있는 3가지 속성(상태 변수)을 가지는 것을 볼수 있다.

      => 즉 Job 객체는 코루틴이 어떤 상태에 있는지 나태내는 위의 3개의 변수들로 외부에 공개한다. (간접적)

  - isActive : 코루틴이 활성화 되어있는지 여부를 나타낸다. 여기서 활성화는 실행 후 취소가 요청되거나 실행이 완료되지 않은 상태를 의미한다.

  - isCanceled : 코루틴 요청이 취소됐는지 여부를 나타낸다. 단 isCanceled가 true 여도 즉시 취소되는 것은 아니다.

  - isCompeleted : 코루틴의 모든 코드가 실행 완료되거나 취소 완료면 true를 반환한다.

  -  코루틴의 상태는 생성, 실행중. 실행완료중, 실행완료, 취소중, 취소 완료 6가지의 상태를 가진다. 

  -  New : 코루틴 빌더를 통해 코루틴 생성시 기본적으로 생성 상태에 놓이며 자동으로 실행중 상태로 넘어간다.

      => 자동으로 변경되지 않도록 하려며 CoroutineStart.Lazy를 사용하여 지연 코루틴을 만들면 된다.

  -  Active : 코루틴이 실행중인 상태를 의미한다. (+ 실행 후에 일시중단 되어도 실행중 상태)

  -  Completing : 부모코루틴의 작업이 모두 끝났는데 자식코루틴의 작업이 끝나지 않은 상태를 의미한다.

  -  Completed : 코루틴의 모든 코드가 실행완료된 경우 실행완료 상태가 된다.

  -  Cancelling : 코루틴에 취소 요청되었을 경우 취소중 상태로 넘어간다. 이때 취소가 된 상태는 아니기 때문에 코루틴은 계속해서 실행된다.

  -  Cancelled : 코루틴의 취소가 확인 된 경우 취소 완료 상태가 된다.

취소요청을 해도 로직이 끝날때까지 실행된다.

 

나.  코루틴 디스패처  

 a.  무엇인가

  -  코루틴 실행을 관리하는 주체로 실행 요청된 코루틴을 스레드로 보내 실행시키는 역할을 한다.

 

 b.  어딘가 익숙한 Dispatcher ? 

  -  코루틴 디스패처 라는 이름을 들으니 Spring의 디스패처 서블릿이 생각나는데 Dispatcher의 큰 특징은 중앙허브 역할로 요청이나 작업을 

     적절한 처리기로 분배하는 역할을 한다.

  - 디스패처 서블릿이 http 요청을 컨트롤러에 분배하는 것 처럼 비슷하게 코루틴 디스패처는 코루틴을 적절한 스레드에 분해하는 역할을 하는 것 같다. 

 

 c.  객체일까?

  -  단순 객체를 의미하는 것인지 궁금해져서 디버깅 해보니 Single Dispatcher 라는 스레드가 하나 생긴 것이 보인다.

      => 즉 작업을 분배 & 실행 시키는 역할을 하는 하나의 스레드이다.

  -  코루틴 디스패처는 내부적으로 작업대기열을 가지고 있으며 코루틴을 실행할 스레드가 없는 경우에는 이 작업대기열에 두고 다른 작업 완료후 실행시킨다.

       => 실행요청이 들어오면 일반적으로 작업대기열에 적재한 후에 스레드로 보내지만 실행 옵션에 따라 즉시실행 될 수 있고 대기열이 없는 것도 있다.  

  -  추가로 코루틴 디스패처는 코루틴을 실행시킬 스레드풀을 생성하고 관리하는 역할도 하고 있다.

 

 d.  코루틴 디스패처 사용해보기

  -  위의 예시처럼 코루틴 디스패처는 코루틴 빌더 함수에 context로 지정해 줄 수 있다.

  -  newSingleThreadContext를 이용해서 코루틴 디스패처를 만들수 있는데 코루틴을 생성해보면 하나의 스레드만 생긴것을 볼 수 있다.

      => 이렇게 코루틴을 실행 시킬때 보낼 수 있는 스레드가 제한된 코루틴 디스패처를 제한된 디스패처라고 부른다.  

  -  newSingleThreadContext가 warning이 뜨는데 이는 특정 디스패처에서만 사용되는 스레드풀이 생성되어 비효율적으로 동작 할수 있기 때문이다.

  -  newFixedThreadPoolContext를 이용하면 코루틴을 실행시키는 스레드를 원하는 개수 만큼 만들어 줄수도 있다.

     => newSingleThreadContext 내부적으로 newFixedThreadPoolContext을 사용하고 있다

     => 즉 해당 함수도 마찬가지로 warning이 발생하는 이유는 위와 동일하며 실제로 사용하는 경우 주의가 필요하다.(쓰지말라는 것 같다..)  

  -  위의 사용한 함수들은 코루틴 디스패처를 직접 만들어서 썼다고 볼수 있는데 코틀린 라이브러리에서 이미 정의된 Dispatcher들을 제공하고 있다.

  -  코루틴 라이브러리에서 미리 만들어둔 코루틴 디스패처 중에 Dispatcher.Default를 먼저 사용해보았다. 

  -  Default는 CPU를 많이 사용하는 연산 즉 CPU 바운드 작업을 할때 사용한다. (ex. 대용량 데이터를 처리하는 작업)

     => CPU를 지속적으로 사용하는 작업을 의미하며 스레드가 지속적으로 사용되기 때문에 코루틴이나 스레드 기반이나 속도차이가 거의 없을 수 있다.

  -  Dispatcher.IO 는 네트워크 요청이나 파일입출력 작업을 위한 디스패처 이다. (ex. http 요청 or DB 작업 등)

  -  스레드 최대 개수는 사용가능한 프로세서의 수와 64개 중 더 작은 값을 사용한다.

  -  Dispatcher.Default 나 Dispatcher.IO 에서 코루틴을 실행시켜보면 같은 스레드풀의 worker 스레드를 사용하는 것을 볼 수 있다

  -  코루틴 라이브러리에서 제공하는 코루틴 디스패처는 라이브러리의 공유 스레드풀을 사용하기 때문에 스레드 생성 & 관리를 효율적으로 할 수 있다.

  -  위의 limitedParallelism로 스레드 수를 제한하는 경우에 Dispatcher.IO는 공유스레드 풀 내에 있는 새로운 스레드풀을 만들어내어서 유의해야한다.

      => 제한없이 limitedParallelism 값을 통해서 스레드를 생성하기 때문에 다른 작업에 영향이 받지 않는 풀을 만들어야 할때만 사용해야한다. 

  -  Dispatcher.Default 의 limitedParallelism 같은 경우에는  Dispatcher.Default가 사용하는 스레드들 중 일부를 사용하는 구조이다.

      =>  Dispatcher.IO와 다르게 warning이 발생 안할 것으로 생각했는데 발생해서 이유를 찾아봐야겠다. 

 

다.  코루틴 처리하기

 a.  순서대로! 

  -  순차처리가 필요한 경우 join을 사용하면 코루틴이 실행완료 될때까지 호출한 코루틴을 일시 중단하도록 할 수 있다.

     => 1번 작업이 반드시 2번 작업 이후에 실행되어야 하는 경우에는 필수적으로 사용해야한다.

  -  주의할점은 join은 호출한 코루틴만을 일시 중지하기 때문에 작업 순서를 잘 고려해서 사용해야 한다

  -  추가로 여러개의 코루틴 작업이 끝나길 기다려야 한다면 joinAll을 사용해서 처리하면 깔끔하다..!

  

 b. 코루틴 자체를 늦게 실행하고 싶다면?

  -  CorotineStart.LAZY를 이용하면 코루틴 자체를 원하는 시점에 동작하도록 하는 것이 가능하다.

 

 d. 코루틴의 취소를 확인하고 싶다면?

  -  cancel을 이용해서 위의 예시처럼 코루틴을 취소할 수 있다. 

  -  위에 cancel만 사용했을 경우에는 완전히 취소가 된 상태가 아닌 취소 중인 상태인 것을 볼 수 있다.

      => flag만 변경된 상태기 때문에 내부적으로 로직이 동작중 일수도 있어서 취소가 완전히 된 이후에 실행이 필요하면 cancelAndJoin를 사용해야한다.

영원히 기다림..

  -  cancel이나 cancelAndJoin 함수를 사용했다고 해서 코루틴 작업이 즉시 취소 되지않는다.

      => 코루틴이 변경된 플래그를 확인하는 시점에 취소가 이루어지기 때문이다.  

  -  코루틴이 취소를 확인하는 시점은 일반적으로 일시중단지점이나 코루틴이 실행을 대기하는 시점이다.

기다리

  -  그렇기 때문에 취소 확인 시점을 만들어 주어야 하는데 3가지 방법이 있다. (명시적으로 취소됐는지 확인하는 코드를 추가하는 방법들)

       1. delay : 일시중단 함수로 선언되어있음 (매번 일시중단이 발생하기 때문에 비효율적)

       2. yield :  호출하는 시점에 코루틴은 자신이 사용하던 스레드를 양보 (매번 일시중단이 발생하기 때문에 비효율적)

       3. CoroutineScope.isActive

  -  어떤 방법이든 체크하는 시점 이전까지는 로직이 동작하는 부분은 주의해야 할것으로 보인다!!

 

라.  async 와 deferred

 a. 작업에 대한 결과가 필요한데.. 

  -  launch로 만든 코루틴은 기본적으로 작업 실행 후 결과를 반환하지 않는데 실무에서는 빈번히 비동기 작업에 대한 결과가 필요하다.

  -  이때 async 코루틴 빌더 함수를 사용하면 코루틴으로 부터 결과값을 수신 받을 수 있도록 한다.

 

 b. Deferred

  -  launch에서는 결과값이 반환되지 않기 때문에 Job 객체를 반환 하는 반면에  async 함수의 반환값을 살펴보면 Deferred라는 것을 볼 수 있다.

     => Deferred 객체는 Job 객체의 특수한 형태로 Deferred 인터페이스는 Job 인터페이스의 서브타입으로 선언된 인터페이스 이다.

  -  Deferred는 Job과 같이 코루틴을 추상화한 객체이지만 코루틴으로부터 생성된 결과값을 감싸는 기능을 추가로 가진다.

  -  미래의 어느 시점에 결과값이 반환될 수 있음을 표현하는데 언제 결과값이 반환될지 정확히 알 수 없어서 필요한 경우에는 수신될때 까지 대기해야 한다.

      => 이때 결과값 수신을 대기하기 위해 await 함수를 제공한다.

  -  await 함수는 코루틴이 실행완료될떄까지 호출부의 코루틴을 일시 중단한다는 점에서 join 함수와 매우 유사하게 동작한다.

     => 예를들어 runBlocking 블록에서 await 함수를 호출하면 runBlocking 코루틴이 일시 중단된다고 보면 된다. 

 

 c. withContext 

  -  withContext 함수를 사용하면 async-await 작업을 대체할 수 있다. 

      => 겉보기에는 async와 await를 연속적으로 호출하는것 같지만 내부적으로 다르게 동작한다.

  -  withContext는 실행중이던 코루틴을 그대로 유지한 채로 코루틴의 실행 환경만 변경해서 작업을 처리한다.

      => 코루틴을 실행시키는 코루틴디스패처 객체가 변경되어 코루틴 실행 스레드가 변화하는 것을 의미한다.

  -  여기서 코루틴의 실행환경(context) 를 변경시키기 때문에 이것도 컨텍스트 스위칭으로 볼 수 있다. 

  -  주의할점은 새로운 코루틴을 만들지 않기 때문에 독립적인 작업을 병렬처리하는 경우 성능문제가 발생할 수 있다.

 

 d.  예시만들기

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis


fun main(): Unit = runBlocking {
    val time = measureTimeMillis {
        val roomId = 10907177L

        val roomInfo = async {
            println("${Thread.currentThread().name} 정보 API 호출")
            getRoomInfoApi(roomId)
        }

        val roomRealTimePrice = async {
            println("${Thread.currentThread().name} 가격 API 호출")
            getRoomPrice(roomId)
        }

        val room = roomInfo.await().apply {
            println("${Thread.currentThread().name} 가격 정보 업데이트")
            updatePrice(roomRealTimePrice.await())
        }

        println(room)
    }

    println("총 시간: $time ms")

}

suspend fun getRoomInfoApi(roomId : Long) : RoomInfo {
    delay(1000L)
    return RoomInfo("1301호")
}

suspend fun getRoomPrice(roomId: Long) : Long  {
    delay(1500L)
    return 550_000L
}

data class RoomInfo(
    private val name: String,
) {
    private var price : Long = 0L

    fun updatePrice(price: Long) {
        this.price = price
    }

    override fun toString(): String {
        return "RoomInfo(name='$name', price=$price)"
    }
}

=== 결과 ===
main 정보 API 호출
main 가격 API 호출
main 가격 정보 업데이트
RoomInfo(name='1301호', price=550000)
총 시간: 1516 ms

  -  async를 이용하여 API를 호출 결과를 받아오는 예시를 만들어 보았다. 위의 코드에서 볼 수 있듯이 API 호출들이 비동기로 처리되는것을 볼 수 있다.

  -  여기서는 메인메서드 하나만으로 비동기 처리를 해보았는데 Thread를 사용하는 비동기 처리 보다 효율적일 것으로 보인다.

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis

fun main(): Unit = runBlocking {
    val time = measureTimeMillis {
        val roomId = 10907177L

        val roomInfo = async {
            println("${Thread.currentThread().name} 정보 API 호출")
            getRoomInfoApi(roomId)
        }.await()

        val roomRealTimePrice = async {
            println("${Thread.currentThread().name} 가격 API 호출")
            getRoomPrice(roomId)
        }.await()

        val room = roomInfo.apply {
            println("${Thread.currentThread().name} 가격 정보 업데이트")
            updatePrice(roomRealTimePrice)
        }

        println(room)
    }

    println("총 시간: $time ms")

}

suspend fun getRoomInfoApi(roomId : Long) : RoomInfo {
    delay(1000L)
    return RoomInfo("1301호")
}

suspend fun getRoomPrice(roomId: Long) : Long  {
    delay(1500L)
    return 550_000L
}

=== 결과 ===
main 정보 API 호출
main 가격 API 호출
main 가격 정보 업데이트
RoomInfo(name='1301호', price=550000)
총 시간: 2520 ms

  -  주의사항은 await을 호출하면 await을 호출한 코루틴에서 값을 받아오기전까지 일시중단이 발생하므로 비동기작업의 결과가 필요한 시점에 호출해야한다.

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis


fun main(): Unit = runBlocking {
    val time = measureTimeMillis {
        val roomId = 10907177L

        val roomInfo = async(Dispatchers.IO) {
            println("${Thread.currentThread().name} 정보 API 호출")
            getRoomInfoApi(roomId)
        }

        val roomRealTimePrice = async(Dispatchers.IO) {
            println("${Thread.currentThread().name} 가격 API 호출")
            getRoomPrice(roomId)
        }

        val room = roomInfo.await().apply {
            println("${Thread.currentThread().name} 가격 정보 업데이트")
            updatePrice(roomRealTimePrice.await())
        }

        println(room)
    }

    println("총 시간 : $time ms")

}

=== 결과 ===
DefaultDispatcher-worker-1 @coroutine#2 정보 API 호출
DefaultDispatcher-worker-3 @coroutine#3 가격 API 호출
main @coroutine#1 가격 정보 업데이트
RoomInfo(name='1301호', price=550000)
총 시간 : 1514 ms

import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis


fun main(): Unit = runBlocking {
    val time = measureTimeMillis {
        val roomId = 10907177L

        val roomInfo = withContext(Dispatchers.IO) {
            println("${Thread.currentThread().name} 정보 API 호출")
            getRoomInfoApi(roomId)
        }

        val roomRealTimePrice = withContext(Dispatchers.IO) {
            println("${Thread.currentThread().name} 가격 API 호출")
            getRoomPrice(roomId)
        }

        val room = roomInfo.apply {
            println("${Thread.currentThread().name} 가격 정보 업데이트")
            updatePrice(roomRealTimePrice)
        }

        println(room)
    }
    println("총 시간: $time ms")
}

=== 결과 ===
DefaultDispatcher-worker-1 @coroutine#1 정보 API 호출
DefaultDispatcher-worker-1 @coroutine#1 가격 API 호출
main @coroutine#1 가격 정보 업데이트
RoomInfo(name='1301호', price=550000)
총 시간: 2523 ms

  -  withContext와 async를 비교하기 위해 같은 예제를 만들어보았는데 async-await 을 바로 썼을 경우와 비슷한 성능을 보이는 것을 볼 수 있다.
  -  추가로 실행된 코루틴이 async에서와 다르게 동일한 것을 볼 수 있다. 즉 withContext는 새로운 코루틴을 생성하지 않는다.