<Coroutine> 4. 구조화된 코루틴

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

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

 

가.  반성..

 a.  오류 발생 재연

val appModule = module {
    single { CoroutineScope(dispatcher) }
}

val dispatcher : CoroutineDispatcher = Executors.newFixedThreadPool(5).asCoroutineDispatcher()

fun Application.configureRouting(){

    val coroutineScope by inject<CoroutineScope>()

    routing {
        get("/api/v1/search") {
            val number = call.parameters["number"]!!.toLong()
            call.respondText(work(coroutineScope = coroutineScope, number = number))
        }
    }
}

suspend fun work(
    coroutineScope: CoroutineScope,
    number: Long
): String {
    println("$coroutineScope")
    println("$number 요청번호")

    val result: String = coroutineScope.async {
        launch {
            println("1번 작업 수행중")
            launch {
                println("2번 작업 수행중")
                throw SQLException("예외 발생!")
            }
        }
        delay(100L)
        "완료"
    }.await()

    return result
}

  -  실제로 코루틴을 한번 적용해보고 싶어서 만들었다가 오류가 발생했던 문제를 재연해 보았다. (개발서버에서만 발생해서 정말 다행이었다..)  

     => Ktor가 궁금해서 요걸로 API를 만들어봤기 때문에 조금 이상해 보일수도 있다...

 

 b.  잘 알지 못하고 적용한 결과

CoroutineScope(coroutineContext=[JobImpl{Active}@681da036, java.util.concurrent.ThreadPoolExecutor@4e8ae368[Running, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]])
1 요청번호
1번 작업 수행중
2번 작업 수행중
2024-06-30 18:05:10.053 [eventLoopGroupProxy-4-1] ERROR ktor.application - Unhandled: GET - /api/v1/search
java.sql.SQLException: 예외 발생!
	at com.example.plugins.RoutingKt$work$result$1$1$1.invokeSuspend(Routing.kt:34)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
CoroutineScope(coroutineContext=[JobImpl{Cancelled}@681da036, java.util.concurrent.ThreadPoolExecutor@4e8ae368[Running, pool size = 4, active threads = 0, queued tasks = 0, completed tasks = 4]])
2 요청번호
2024-06-30 18:05:12.012 [eventLoopGroupProxy-4-2] DEBUG ktor.application - Unhandled: GET - /api/v1/search. Exception class kotlinx.coroutines.JobCancellationException: Parent job is Cancelled]
kotlinx.coroutines.JobCancellationException: Parent job is Cancelled
Caused by: java.sql.SQLException: 예외 발생!
	at com.example.plugins.RoutingKt$work$result$1$1$1.invokeSuspend(Routing.kt:34)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)

  -  공통 적용을 하고 싶어서 CoroutineScope를 Bean으로 만들었는데 그 안에 만들었던 launch에서 예외가 발생했고 결과로CoroutineScope 자체가

     취소완료 처리되어 만약에 정상적인 결과를 보낼수 있음에도 독립적인 http 요청에도 자식 코루틴 실행없이  JobCancellationExcpetion이 발생하였다.

     =>  트래픽이 많은 API 였다면 정말 대형 장애로 발생할 수 있었을 수도 있는 오류였다는 생각도 들었다.. 

  -  책을 읽고 나니까 코루틴에 대한 예외처리가 되지 않았고 CoroutineScope를 Bean으로 사용한 것이 문제라고 생각되었다. 

     => 만약에 여러개 API에서 CoroutineScope를 공유한다면 1개의 API에서만 예외가 발생해 취소완료처리 된다면 모든 API를 사용할수 없게 된다.

 

c.  느낀점

  -  어드민에 적용하는거라서 가볍게 생각했던 부분이 있었는데 이 챕터를 읽으면서 잘못 만들었다는 걸 정확하게 알 수 있었다. 

  -  정말 잘 알지 못한 상태에서 사용법만 가지고 만들었기 때문에 이런 문제가 발생했다.. 항상 잘알고 써야겠다는 생각이 다시 한번 더 드는것 같다!

 

나.  코루틴 구조

 a. 구조화된 동시성

  -  비동기 작업을 구조화해서 비동기 처리를 보다 안정적이고 예측할 수 있게 만드는 것을 구조화된 동시성 원칙이라고 한다.

  -  코루틴의 경우 부모-자식 관계로 구조화해서 보다 안전하게 관리되고 제어될 수 있도록 하고 있다.

 

 b. 구조화된 코루틴 특징

  -  부모 코루틴의 실행환경(CoroutineContext)이 자식코루틴에게 상속 된다.

  -  부모코리틴은 자식 코루틴이 완료될때까지 대기한다.

  -  CoroutineScope를 사용해 코루틴이 실행되는 범위를 제한 할수 있다.

 

 c.  실행 환경 상속

  -  위의 예시의 결과처럼 코루틴은 별개로 생성되었지만 실행환경은 상속되는 것을 볼 수 있고 별도로 다르게 줄 수도 있는 것을 알 수 있다.

  -  주의할 점은 코루틴빌더는 사용할때마다 Job객체를 새롭게 생성하는데 이를 상속하게 되면 제어가 어려워지기 때문에 Job객체는 상속되지않는다.

 

다.  코루틴 작업제어

 a.  Job 객체는 부모-자식 형태와 관련이 없을까?

   -  위에서 Job객체를 상속하지는 않지만 Job객체는 가장 중요한 코루틴을 구조화 하는데 사용된다.

  - Job의 인터페이스를 살펴보면 child와 parent를 프로퍼티로 가지고 있는 것을 볼 수 있다.

     => 이를 통해서 부모 자식 코루틴간의 양방향 참조를 가지고 제어할 것이라는 것을 추측해 볼 수 있다.

 -  코루틴은 하나의 부모 코루틴만을 가지며 Job?를 통해 최상위 코루틴은 부모 코루틴이 없기 때문에 null 을 가진다. 자식은 여러개를 가질 수 있다.

 

 b. 부모-자식 전파

  -  코루틴은 자식 코루틴으로 취소를 전파하는 특성을 갖기 떄문에 특정 코루틴이 취소 되면 하위의 모든 코루틴이 취소된다.

      => 특정 코루틴에 취소가 요청 되면 취소는 자식 코루틴 방향으로만 전파되고 부모 코루틴으로는 전파되지 않는다.

  -  모든 자식코루틴이 끝나야 부모코루틴이 끝나는 이유는 큰 작업을 연관된 여러 작은 작업으로 나누는 방식이기 때문이다.

  -  Job의 invokeOnCompletion 함수를 사용하면 Job이 완료 or 취소로 끝날때 콜백 로직을 설정해줄 수 있다.

 

 c. 실행완료중 상태(Completing)

  -  부모코루틴의 모든 작업이 실행되었지만 자식코루틴중 하나라도 실행중인 경우에 부모코루틴은 실행완료중 상태를 가진다.

      => 상태값은 실행완료와 동일하게 isActive=true, isCompleted=false 인 상태이다.

 

라.  코루틴 스코프(CoroutineScope)

 a.  무엇인가?

  -  CoroutineScope는 코루틴들에게 CoroutineContext를 제공하고 이들의 범위 & 관리하는 역할을 한다.

  -  위의 이미지에서 보면 확장함수를 제외하고 단순 코루틴 컨텍스트만을 프로퍼티로 가진 인터페이스인 것을 볼 수 있다.

 

 b.  단순 CoroutineContext  사용할때와 비교해보자

fun main(): Unit = runBlocking {

    val context = Dispatchers.Default + CoroutineName("Hello")

    val job = launch(context = context) {
        println("부모코루틴 ${Thread.currentThread().name}")

        launch {
            delay(1000L)
            println("자식코루틴 ${Thread.currentThread().name}")
        }
    }
    job.invokeOnCompletion {
        println("작업완료")
    }
}

=== 결과 ===
부모코루틴 DefaultDispatcher-worker-1 @Hello#2
자식코루틴 DefaultDispatcher-worker-2 @Hello#3
작업완료

fun main2 (): Unit = runBlocking {

    val context = Dispatchers.Default + CoroutineName("Hello")
    val coroutineScope = CoroutineScope(context)

    val job = coroutineScope.launch(context = context) {
        println("부모코루틴 ${Thread.currentThread().name}")

        launch {
            delay(1000L)
            println("자식코루틴 ${Thread.currentThread().name}")
        }
    }
    job.invokeOnCompletion {
        println("작업완료")
    }
}

=== 결과 ===
부모코루틴 DefaultDispatcher-worker-1 @Hello#2

  -  CorotineScope를 별도로 생성하고 해당 Scope에서 코루틴을 만든경우 모든 코루틴이 실행되기 전에 프로세스가 종료 되었다.

  -  예시를 만들어 자세히 확인해보면 별도 CoroutineScope로 만든 코루틴은 runBlocking의 자식코루틴으로 포함되지 않는것을 확인 할 수 있었다.

     =>  즉 CoroutineScope 함수를 사용해 새롭게 생성하면 기존 계층구조를 따르지 않는 새로운 Job 객체가 생성된다.

 -  실제로 코루틴의 범위와 범위를 만드는 것은 Job 객체이고 이는 CoroutineScope 객체에 의해 관리되는 것이다.

 -  코루틴의 구조화를 깨는 것은 비동기 작업을 안전하지 않게 만들기 때문에 최대한 지양해야한다.

 

 c.  상세

  -  CoroutineScope 내부에서 실행되는 코루틴이 CoroutineScope 부터 CoroutineContext를 제공하는 것을 확인했다.

  1. 수신객체인 CoroutineScope로부터 CoroutineContext를 제공받는다

  2. 제공받은 CoroutineContext + launch 함수의 파라미터로 넘어온 context를 더해 새로운 CoroutineContext를 만든다.

  3. 생성된 CoroutineContext에 코루틴 빌더 함수가 호출되어 새로 생성되는 Job을 더한다. (이때 전달되는 Job 객체가 있으면 부모 Job이 된다.)

  -  앞서서 launch함수에서 this로 coroutineContext에 접근할수 있었던 이유는 CoroutineScope가 제공 했기 때문이다. (람다식 내의 코루틴을 모두 포함)

     => 마찬가지로 자식-부모 관계 역시  CoroutineScope로 부터 제공되어 상속했기 때문에 구조화된 코루틴을 사용할 수 있다.

 

 d. 구조화와 취소 전파

import kotlinx.coroutines.*

fun main(): Unit = runBlocking {

   val rootJob = launch {
        println("1번 작업 시작")

        val parent1 = launch {
            println("1-1번 작업 시작")

            launch {
                println("1-1-1번 작업 시작")
                delay(1000L)
                println("1-1-1번 작업 종료")
            }

            launch {
                println("1-1-2번 작업 시작")
                delay(1000L)
                println("1-1-2번 작업 종료")
            }

            println("1-1번 작업 종료")
        }

        launch {
            println("1-2번 작업 시작")
            launch {
                println("1-1-3번 작업 시작")
                delay(1000L)
                println("1-1-3번 작업 종료")
            }
            println("1-2번 작업 종료")
        }

        println("1번 작업 종료")
        parent1.cancel()
    }
    // rootJob.cancel() -> 아무것도 출력하지 않는다  
}

=== 결과 ===
1번 작업 시작
1번 작업 종료
1-2번 작업 시작
1-2번 작업 종료
1-1-3번 작업 시작
1-1-3번 작업 종료

  -  부모 코루틴의 Job이 취소가 되면 자식 코루틴들에게 취소가 전파되어 일반적으로 부모코루틴 하위의 모든 자식코루틴들도 작업이 취소 된다.

import kotlinx.coroutines.*

fun main(): Unit = runBlocking {

   val rootJob = launch {
        println("1번 작업 시작")

        CoroutineScope(Dispatchers.Default).launch {
            delay(1000L)

            println("1-1번 작업 시작")
            launch {
                println("1-1-1번 작업 시작")
                delay(1000L)
                println("1-1-1번 작업 종료")
            }

            launch {
                println("1-1-2번 작업 시작")
                delay(1000L)
                println("1-1-2번 작업 종료")
            }

            println("1-1번 작업 종료")
        }

        launch {
            delay(1000L)

            println("1-2번 작업 시작")
            launch {
                println("1-1-3번 작업 시작")
                delay(1000L)
                println("1-1-3번 작업 종료")
            }

            println("1-2번 작업 종료")
        }

        println("1번 작업 종료")
    }
    delay(1L)
    rootJob.cancel()
    delay(3000L)
}

=== 결과 ===
1번 작업 시작
1번 작업 종료
1-1번 작업 시작
1-1-1번 작업 시작
1-1번 작업 종료
1-1-2번 작업 시작
1-1-2번 작업 종료
1-1-1번 작업 종료

  -  CoroutineScope 를 이용해서 코루틴의 구조화를 깨보면 취소가 전파되지 않는 것을 볼 수 있다. 다른 방법으로는 context에 Job 객체를 추가하면 된다.

      =>  즉 작업에 대한 관리는 Job 객체가 하고 있고 CoroutineScope도 새로운 Job 객체를 생성했기 때문에 구조화가 깨지는 걸로 볼 수 있다.

 

 d. 주의 : 새로 만든 Job은 자동으로 실행완료 되지 않는다. 

import kotlinx.coroutines.*

fun main(): Unit = runBlocking {

   val job = this.coroutineContext[Job]
   val newJob = Job(parent = job)

    launch(context = newJob) {
        println("1번 작업 시작")
        delay(1000L)
        println("1번 작업 종료")

    }.invokeOnCompletion {
        println("코루틴 실행완료")
    }

    newJob.invokeOnCompletion {
        println("생성된 Job 실행 완료")
    }

}

=== 결과 ===
1번 작업 시작
1번 작업 종료
코루틴 실행완료

  -  위의 코드는 프로세스가 종료가 되어지지 않는데 그이유는 새로 만든 newJob이 완료 처리가 되지 않았기 때문이다.

      =>  단순 코드만 보면 newJob 객체가 1번 작업에 대한 코루틴의 Job 객체로 보이지만 사실은 자식 코루틴에서 실행되는 것이다.

  -  명시적으로 .complete() 함수를 호출해야 정상적으로 종료가 된다.

 

 e.  runBlocking과 launch의 차이 

  -  runBlocking은 BlockingCoroutine 을 만들고  launch는 일반적으로 StandaloneCoroutine 을 만든다  

      => 둘다 코루틴을 생성하는 코루틴 빌더 함수 이지만 호출부의 스레드를 사용하는 방법에 차이가 있다.

  -  runBlocking으로 만들어진 BlockingCoroutine은 실행이 완료될때까지 호출부의 스레드를 차단하고 사용한다.

      => 이 코루틴의 실행이 완료 될때까지 호출부의 스레드가 다른 작업에 사용될 수 없다.(BlockingCoroutine 의해 베타적으로 사용됨)

  -  이 차단은 스레드 블로킹과는 다른데 스레드 블로킹은 어떤 작업에도 사용할 수 없도록 차단되는 것이지만  runBlocking 함수의

      차단은 자신의 코루틴과 그 자식 코루틴을 제외한 다른작업이 스레드를 사용할 수 없는 부분에서 차이가 있다.

  -  runBlocking은 블로킹을 일으키는 일반코드와 코루틴 사이의 연결점 역할을 하기 위해 만들어서 코루틴 내부에서 다시 또 호출하는 것은 피해야한다.

  -  반면에 launch 함수로 만들어진 코루틴은 호출부의 스레드를 차단하지 않는다는 차이점을 가지고 있다.