Kotlin - coroutine!

코루틴

참고: https://myungpyo.medium.com/reading-coroutine-official-guide-thoroughly-part-0-20176d431e9d

코루틴은 Co(together)+routine 의 합성어, 상호 연계 프로그램을 일컫는다, 다른 언어들에도 많은 라이브러리들이 코루틴 기능을 지원한다.
별도의 스레드 관련 코드없이 프로세스의 동시적 실행을 지원하는 협력형 멀티태스킹 개념을 사용한다.

kotlin 에서도 코루틴 기능을 적극 지원하며 많은 개발자들이 활발히 사용중이다.
간단한 코루틴 예제를 통해 동시성 실행되는것을 확인할 수 있다.

fun task1() {
    println("start task1 in Thread ${Thread.currentThread()}")
    println("end task1 in Thread ${Thread.currentThread()}")
}

fun task2() {
    println("start task2 in Thread ${Thread.currentThread()}")
    println("end task2 in Thread ${Thread.currentThread()}")
}

println("start")

kotlin.run {
    task1()
    task2()
    println("called task1 and task2 from ${Thread.currentThread()}")
}
println("done")
/* 
start
called task1 and task2 from Thread[main,5,main]
start task1 in Thread Thread[main,5,main]
start task2 in Thread Thread[main,5,main]
end task1 in Thread Thread[main,5,main]
end task2 in Thread Thread[main,5,main]
done, 1031
*/

출력값과 실행시간을 보면 run 내부 함수들은 동시실행 되었으며, 코드노이즈 부분에서 js 의 callback method, java 의 reactive stream 과 비교했을 때 더 간결한 코드작성이 가능하다.

각 메서드의 콜스택이 main 메서드 코드 사이사이에 인터리브되어 모든 함수가 main 스레드 에서 실행됨을 알 수 있다.

suspend 함수 (일시중단함수)

suspend 키워드는 시작하고 멈추고 다시 시작할 수 있는 함수로 비동기 실행을 위한 중단 지점(suspention point) 을 의미하는 키워드이다.

I/O 처리나 외부 서비스와의 연계등 중단될만한 지점이 있다면 suspend 함수 로 정의해두고 내부에 동시성 코드를 작성하면 된다.

kotlin 에서 제공하는 대표적인 suspend 함수로 [delay, yield] 가 있다.
suspend 함수를 맞닥뜨리면 스레드는 작업이 완료될 동안 다른작업을 수행한다.
또한 suspend 함수 는 항상 suspend 함수 내부에서 사용되어야 한다.

delay 는 지정된 밀리초만큼 실행을 정지시킨다.
간단히 스레드 하나짜리 스레드풀과 suspend 함수를 테스트하면 아래와 같다.

// 스레드 non-block suspend 메서드
suspend fun suspendReturnOne(): Int {
    println("suspendReturnOne ${Thread.currentThread().name}")
    delay(3000L)
    return 1
}

suspend fun suspendReturnTwo(): Int {
    println("suspendReturnTwo ${Thread.currentThread().name}")
    delay(1000L)
    return 2
}

fun main() {
    val singleThreadDispatcher = newSingleThreadContext("Single Thread ThreadPool")
    var time = measureTimeMillis {
        runBlocking(singleThreadDispatcher) {
            // 병렬로 2개 함수 실행
            var result1 = async { suspendReturnOne() }
            var result2 = async { suspendReturnTwo() }
            println(result1.await() + result2.await())
        }
    }
    println("time: $time")
}
/*
suspendReturnOne Single Thread ThreadPool
suspendReturnTwo Single Thread ThreadPool
3
time: 3038
*/

싱글스레드임에도 불구하고 suspend 키워드를 사용해 suspend 함수 를 정의하였고 delay 를 만나는 순간 바로 다른 작업을 하러가기 때문에 총 3초를 소요하게된다.

yield 는 별도의 명시된 시간 없이 실행을 정지시킨다. 다른언어의 yield 와 마찬가지로 stop interrupt를 발생시키며 스레드의 흐름을 제어한다.

suspend fun task1() {
    println("start task1 in Thread ${Thread.currentThread()}")
    yield()
    println("end task1 in Thread ${Thread.currentThread()}")
}

suspend fun task2() {
    println("start task2 in Thread ${Thread.currentThread()}")
    yield()
    println("end task2 in Thread ${Thread.currentThread()}")
}

fun main() {
    println("start")
    runBlocking {
        launch { task1() }
        launch { task2() }
        println("called task1 and task2 from ${Thread.currentThread()}")
    }
    println("done")
}
/*
start
called task1 and task2 from Thread[main,5,main]
start task1 in Thread Thread[main,5,main]
start task2 in Thread Thread[main,5,main]
end task1 in Thread Thread[main,5,main]
end task2 in Thread Thread[main,5,main]
done
*/

다른 코루틴 중단함수에서 yield 를 호출해주거나 중단함수가 종료되면 다시 제어를 되찾는다.

코루틴 Context

모든 코루틴 동작에는 코루틴 Context 가 존재한다.
코루틴을 식별, 실행하기 위한 다양한 정보들이 코루틴 Context 에 포함되어 있다.

가장 간단한 코루틴 Context 는 Coroutine Id 만 존재하며, 여러가지 요소를 추가해 결합된 코루틴 Context 정의가 가능하다.

아래와 같이 코루틴 Context 와 사용할 Dispatchers 요소를 결합하여 코루틴이 어떤 스레드풀에서 동작할지 결정할 수 있다.

// Coroutine Id 까지 출력시키기 위해 -Dkotlinx.coroutines.debug 옵션 추가  
fun main() {
    println("start")
    runBlocking {
        launch(Dispatchers.Default) { task1() }
        launch { task2() }
        println("called task1 and task2 from ${Thread.currentThread()}")
    }
    println("done")
}
/*
start
start task1 in Thread Thread[DefaultDispatcher-worker-1 @coroutine#2,5,main]
called task1 and task2 from Thread[main @coroutine#1,5,main]
end task1 in Thread Thread[DefaultDispatcher-worker-1 @coroutine#2,5,main]
start task2 in Thread Thread[main @coroutine#3,5,main]
end task2 in Thread Thread[main @coroutine#3,5,main]
done
*/

Coroutine ID 가 [coroutine#1, coroutine#2, coroutine#3] 출력되는것으로 보아 전부 다른 코루틴 Context 에서 수행된는 것을 알 수 있다.

runBlocking 과 launch(Dispatchers.Default) 의 경우 실행되는 스레드가 달라 병렬실행 된다.
runBlocking 과 그냥 launch 의 경우 실행되는 스레드가 같아 동시실행 된다.

위 예제처럼 Dispatcher 요소만 결합하면 알아서 코루틴을 스레드에 따라 코드를 병렬실행 혹은 동시실행 할지 결정할 수 있다.

• Dispatchers.Default: 기본 스레드풀, 일반적인 CPU 작업을 수행하는 코루틴에 사용
• Dispatchers.IO: IO작업 실행을 위한 코루틴에 사용
• Dispatchers.Main: 안드로이드, Swing UI 를 구성 코루틴에 사용

만약 커스텀한 Dispatcher 를 결합하고 싶다면 ExecutorService 로부터 스레드풀을 직접 만들어 사용가능하다.

val dispatcher: ExecutorCoroutineDispatcher = Executors.newFixedThreadPool(10)
    .asCoroutineDispatcher()
...
...
dispatcher.close()
// close 함수를 호출하지 않으면 메인함수가 종료되지 않음으로 호출 필수

ExecutorService 를 패키징한 코루틴 표준 라이브러리 함수도 있다.

val singleThreadDispatcher = newSingleThreadContext("Single Thread ThreadPool")
val multiThreadDispatcher = newFixedThreadPoolContext(10, "Multi Thread ThreadTool")

동시실행, 병렬실행에 대한 설정이 복잡한 스레드관련 코드 없이 코루틴 Context 결합만으로 쉽게 구축가능한것이 kotlin 코루틴 의 장점이다.

Continuation

중단지점으로 인해 동작하는 스레드가 변경되어도 동일한 코루틴 Context 가 유지되는 기능을 Continuation 이라 한다.

suspend 함수는 중단되었다가도 언제든 다시 실행될 수 있도록 하기위해 문맥정보를 유지하는 Continuation 기능을 지원한다.

task(2) 메서드의 출력값을 보면 코루틴 Context 이름은 동일하지만 실행되는 스레드는 다른것을 확인할 수 있다.

suspend fun task(n: Long): Long {
    val factor = 2
    println("start $n task in Thread ${Thread.currentThread()}")
    delay(n * 1000)
    println("end $n task in Thread ${Thread.currentThread()}")
    return n * factor
}

fun main() {
    runBlocking {
        val compute = Compute()
        launch(Dispatchers.Default) {
            task(1)
        }
        launch(Dispatchers.Default) {
            task(2)
        }
    }
}
/*
start 1 task in Thread Thread[DefaultDispatcher-worker-1 @coroutine#2,5,main]
start 2 task in Thread Thread[DefaultDispatcher-worker-2 @coroutine#3,5,main]
end 1 task in Thread Thread[DefaultDispatcher-worker-1 @coroutine#2,5,main]
end 2 task in Thread Thread[DefaultDispatcher-worker-1 @coroutine#3,5,main]
*/

스레드간 코루틴 Context 를 유지하기 위해 변환된 바이트코드를 보면 굉장히 복잡하다.

continuation 을 유지하기 위해 외부에서 문맥정보에 해당하는 참조객체를 매개변수로 전달하고,
스레드풀에서 실행할 수 있는 코루틴 코드를 가진 익명객체를 사용한다.

java 에선 구현이 가능할까 싶다할 코루틴 기능을 kotlin 에서 쉽게 제공한다는 점이 중요하다.

코루틴 Scope

코루틴 Context 가 스레드가 사용하는 코루틴 실행 문맥정보라면,
코루틴 Scope 는 스레드가 코루틴 Context 정보를 가지고 실행시킬 코드영역, 코루틴 Context 가 영향을 끼치는 영역이라 할 수 있다.

public interface CoroutineScope {
    /**
     * The context of this scope.
     * Context is encapsulated by the scope and used for implementation of coroutine builders that are extensions on the scope.
     * Accessing this property in general code is not recommended for any purposes except accessing the [Job] instance for advanced usages.
     *
     * By convention, should contain an instance of a [job][Job] to enforce structured concurrency.
     */
    public val coroutineContext: CoroutineContext
}

코루틴 Context 는 계층형태로 이루어져 있으며, 각 실행되는 코루틴 Context 는 서로 다르더라도 부모-자식 관계를 가지며 최상위 코루틴 Scope 안에서 동작된다.

좀더 자세한 그림은 아래와 같다.

생성된 코루틴 Scope 로부터 코루틴을 정의하고, 그 안에 또 새로운 코루틴을 정의했다.

코루틴 생성시 별도의 코루틴 Context 설정이 없다면 부모 코루틴 Context 로부터 그대로 속성값을 이어받음

미리정의된 코루틴 Scope 를 가져오거나 새로운 코루틴 Scope 를 생성할 수 있다.

fun main(args: Array<String>) {
    // 이미 존재하는 GlobalScope 가져오기
    val existScope = GlobalScope
        .launch {
            print("foo, ")
            delay(4000L)
            print("bar, ")
        }

    // 새로운 코루틴 Scope 생성
    var newScopeJob = CoroutineScope(Dispatchers.Default)
        .launch {
            print("Hello, ")
            delay(2000L)
            print("World, ")
        }
    runBlocking {
        existScope.join()
        newScopeJob.join()
    }
}

그리고 코루틴 Scope 를 사용해 코루틴 블럭 을 정의할 수 있는 코루틴 Builder 메서드들이 존재한다.

생성된 코루틴 Scope 의 확장함수로 코루틴을 동작시킨다.

• launch
• async
• withContext
• coroutineScope

또한 runBlokcing 과 같은 코루틴 Scope 를 사용할 뿐인 일반함수들도 있다.

runBlocking

runBlocking 메서드는 검증 또는 테스트에 사용한다, 현재 스레드를 모든 내부 코루틴이 종료될 때 까지 블록한다.
GlobalScope 코루틴 Scope 의 하위 코루틴 Context 로 동작하며 스레드를 블록하기 때문에 실제 운용시에는 자주 사용하지 않는다.

public fun <T> runBlocking(
    context: CoroutineContext = EmptyCoroutineContext, 
    block: suspend CoroutineScope.() -> T
): T { ... }
// 마지막 표현식 반환

코루틴 Scope 생성 메서드에서 코루틴 Context 를 별도로 지정하지 않으면 EmptyCoroutineContext 를 사용하는데 GlobalScope 에서도 사용하는 싱글톤 코루틴 Context 객체이다. 애플리케이션 프로세스와 동일한 생명주기를 갖는다.

fun main() {
    println("Before creating coroutine")
    runBlocking {
        print("Hello, ")
        delay(4000L)
        println("World!")
    }
    println("After coroutine is finished")
}
/* 
Before creating coroutine
Hello, World!
After coroutine is finished
*/

Hello, 문자열이 출력되고 4초뒤에 이후 문자열들이 출력된다

launch

launch 는 CoroutineScope 인터페이스의 확장함수로 CoroutineScope 내부에서만 사용할 수 있다.

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job
// Job 객체 반환

마지막 block 매개변수로 CoroutineScope 의 확장함수로 구현된 람다 메서드를 전달받는데 해당 람다가 새로운 코루틴으로 동작하게 된다.

launch 메서드가 사용된 CoroutineScope 의 확장함수이니 동알한 CoroutineScope 에서 동작하게 된다.

fun main() {
    println("Before runBlocking")
    runBlocking {
        println("Before launch")
        launch {
            delay(2000L)
            println("Hello, World!")
        }
        println("After launch") // After launch 가 먼저 찍힌것을 확인
    }
    println("After runBlocking")
}
/* 
Before runBlocking
Before launch
After launch
Hello, World!
After runBlocking
*/

표현식을 반환하지 않는 대부분의 코루틴 Builder 메서드들이 Job 인스턴스를 구현한 객체를 반환하는데 코루틴의 취소, 완료대기 등에 사용된다.

withTimeout, withTimeoutOrNull 등의 함수또한 Job 인스턴스를 반환한다.

async

launch 와 마찬가지로 CoroutineScope 인터페이스의 확장함수이다.
async 는 표현식을 사용해 값을 반환해야 하는 경우에 사용한다.

public fun <T> CoroutineScope.async(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> T
): Deferred<T>
// Deferred 객체 반환

Deferred 객체를 반환, java8 Future 와 비슷한 클래스로 Job 의 구현체이다.

await 메서드를 통해 최종값을 가져올때 까지 스레드를 블록한다.

suspend fun add(x: Int, y: Int): Int {
    delay(1000L)
    return x + y
}

fun main() {
    var time = measureTimeMillis {
        runBlocking {
            val firstSum: Deferred<Int> = async {
                println(Thread.currentThread().name)
                add(2, 2)
            }
            val secondSum: Deferred<Int> = async {
                println(Thread.currentThread().name)
                add(3, 4)
            }
            println("Awaiting concurrent sums...")
            val total = firstSum.await() + secondSum.await()
            println("Total is $total")
        }
    }
    println("time=$time")
}
/*
Awaiting concurrent sums...
main @coroutine#2
main @coroutine#3
Total is 11
time=1044
*/

withContext

메서드 이름처럼 코루틴 Context 를 유지하면서 새로운 코루틴을 생성하고 싶을 때 사용한다.

public suspend fun <T> withContext(
    context: CoroutineContext,
    block: suspend CoroutineScope.() -> T
): T

아래와 같이 2개의 withContext 로 생성한 코루틴에 서로 다른 스레드풀을 지정하고 실행시킨다.

suspend fun task(taskName: String, time: Long) {
    println("$taskName start in Thread ${Thread.currentThread()}")
    delay(time)
    println("$taskName end in Thread ${Thread.currentThread()}")
}

fun main() {
    println("start")
    val time = measureTimeMillis {
        runBlocking {
            println("called coroutine context ${Thread.currentThread()}")
            withContext(Dispatchers.Default) { task("first", 1000) }
            withContext(coroutineContext) { task("second", 1000) }
        }
    }
    println("done, time: $time")
}
/*
start
called coroutine context Thread[main @coroutine#1,5,main]
first start in Thread Thread[DefaultDispatcher-worker-1 @coroutine#1,5,main]
first end in Thread Thread[DefaultDispatcher-worker-1 @coroutine#1,5,main]
second start in Thread Thread[main @coroutine#1,5,main]
second end in Thread Thread[main @coroutine#1,5,main]
done, time: 2056
*/

출력처럼 실행되는 스레드풀은 다르지만 동일한 코루틴 Context 이름을 가진다.
그리고 동일한 코루틴 Context 를 유지하기 위해 동기성으로 코루틴 Scope 를 실행 시킨다.

하나의 스레드에서 실행하기 무거운 작업의 경우 스레드풀을 지정해서 실행하면 전체적인 시스템 부하를 줄일 수 있다.

coroutineScope

주어진 코루틴 Scope 와 코루틴 Context 를 이어받아 코루틴을 생성하는 함수

public suspend fun <R> coroutineScope(block: suspend CoroutineScope.() -> R): R =
    suspendCoroutineUninterceptedOrReturn { uCont ->
        val coroutine = ScopeCoroutine(uCont.context, uCont)
        coroutine.startUndispatchedOrReturn(coroutine, block)
    }

기존 스레드와 코루틴 Context 을 그대로 이어서 실행함으로 완료될 때까지 일시중단 된다.

suspend fun testPrint() {
    coroutineScope {
        delay(1000)
        println("testPrint thread ${Thread.currentThread()}")
    }
}

fun main() {
    var time = measureTimeMillis {
        runBlocking {
            testPrint() // testPrint thread Thread[main @coroutine#1,5,main]
            testPrint() // testPrint thread Thread[main @coroutine#1,5,main]
            testPrint() // testPrint thread Thread[main @coroutine#1,5,main]
        }
    }
    println("done: $time") // done: 3031
}

아래와 같이 새로운 자식 코루틴 Context 에서 실행될 수 있도록 launch, async 함수로 감싸는것이 일반적이다.

fun main() {
    var time = measureTimeMillis {
        runBlocking {
            launch { testPrint() } // testPrint thread Thread[main @coroutine#2,5,main]
            launch { testPrint() } // testPrint thread Thread[main @coroutine#3,5,main]
            launch { testPrint() } // testPrint thread Thread[main @coroutine#4,5,main]
        }
    }
    println("done: $time") // done: 1032
}

단순히 코루틴을 suspend 함수로 감싸는 기능을 제공하며 코루틴 코드를 함수화 하고 분리하기 위해서 사용한다.

suspend fun testPrint() {
    coroutineScope {
        launch {
            delay(1000)
            println("hello world $coroutineContext")
        }
    }
}

코루틴 취소

launch, async 는 각각 Job, Deferred 객체를 반환하며 두 객체 모두 cancel, cancelAndJoin 메소드를 이용해 진행중인 코루틴의 취소가 가능하다.

진행중인 코루틴이 바로 취소되진 않고 [yield, delay, await] 와 같은 중단 지점(suspention point) 을 만났을 때 취소된다.

취소시 중단지점에서 CancellationException 이 발생한다.
중단지점이 없는 코루틴의 경우 실행중 취소된다 하더라도 예외가 발생하지 않아 코드가 중단되지 않는다.

만약 오랜시간이 걸리는 작업을 도중에 취소하는 로직을 넣고싶다면, 코드 사이사이에 yield 와 같은 중단지점을 추가하는 것을 권장한다.

fun main() {
    runBlocking {
        val job = launch {
            try {
                repeat(10) { i ->
                    println("Job: Working $i, active $isActive")
                    delay(500L) // 중단지점
                }
            } catch (e: Exception) {
                println("예외처리 class:${e.javaClass.simpleName}, msg:${e.message}, active:$isActive")
            }
        }
        launch {
            delay(1300L)
            if (job.isActive)
                job.cancel("cancel test!!!")
        }
    }
}
/* 
Job: Working 0, active true
Job: Working 1, active true
Job: Working 2, active true
예외처리 class:CancellationException, msg:cancel test!!!, active:false
*/

부모 코루틴을 취소하면 모든 자식 코루틴이 취소된다.

fun main() {
    runBlocking {
        var outerJob1 = launch {
            println("outerJob1 start")
            var innerJob1 = launch {
                println("innerJob1 start")
                repeat(1000) { i ->
                    delay(1000)
                    println("innerJob1 $i")
                }
            }

            var innerJob2 = launch {
                println("innerJob2 start")
                repeat(1000) { i ->
                    delay(1000)
                    println("innerJob2 $i")
                }
            }
            repeat(1000) { i ->
                delay(1000)
                println("outerJob1 $i")
            }
        }
        delay(3500)
        println("job cancel")
        outerJob1.cancel()
    }
}
/* 
outerJob1 start
innerJob1 start
innerJob2 start
outerJob1 0
innerJob1 0
innerJob2 0
outerJob1 1
innerJob1 1
innerJob2 1
outerJob1 2
innerJob1 2
innerJob2 2
job cancel
*/

방해금지

만약 중단지점이 있음에도 해당 영역이 중요한 연산지점이라 cancel 영향을 무시하고 싶다면 아래와 같이 withContext(NonCancellable) 로 감싼다.

fun main() {
    runBlocking {
        val job = launch {
            try {
                withContext(NonCancellable) {
                    repeat(10) { i ->
                        println("Job: Working $i, active $isActive")
                        delay(500L)
                    }
                }
            } catch (e: Exception) {
                println("예외처리 class:${e.javaClass.simpleName}, msg:${e.message}, active:$isActive")
            }
        }
        launch {
            delay(1300L)
            if (job.isActive)
                job.cancel()
        }
    }
}
/* 
Job: Working 0, active true
Job: Working 1, active true
... 끝까지 수행됨
Job: Working 8, active true
Job: Working 9, active true
*/

withContext(NonCancellable) 로 인해 try catch 까지 예외가 전파되지 않는다.

취소위임

코루틴 Context 내부에는 isActive 속성이 존재하며 Job, Deferred 객체에서 cancel 메서드를 통해 isActive 속성을 변경할 수 있다.

아래와 같이 실시간으로 isActive 속성을 검사하는 방식을 사용하면 중단지점이 없더라도 코루틴 취소 기능을 내부 코드에 위임시켜 while 무한루프에서 탈출할 수 있다.

suspend fun compute(checkActive: Boolean) {
    coroutineScope {
        var count = 0L
        val max = 100_000_000_000L
        while (if (checkActive) isActive else count < max) {
            count++
        }
        if (count == max)
            println("compute, checkActive $checkActive ignored cancellation, count: $count, time: ${System.currentTimeMillis()}")
        else
            println("compute, checkActive $checkActive bailed out early, count: $count, time: ${System.currentTimeMillis()}")
    }
}

fun main() {
    runBlocking {
        val job1 = launch (Dispatchers.Default) { compute(checkActive = true) }
        val job2 = launch (Dispatchers.Default) { compute(checkActive = false) }
        if (job2.isActive) job2.cancel()
        if (job1.isActive) job1.cancel()
    }
    println("done")
}
/* 
compute, checkActive true bailed out early, count: 1136, time: 1696482800089
compute, checkActive false ignored cancellation, count: 100000000000, time: 1696482831424
done
*/

결과값을 보면 checkActive=true 의 경우 max 를 모두 채우지 않고 빠져나온것을 알 수 있다.
checkActive=false 의 경우 30초가량 수행되었음에도 중단지점이 없어 도중 취소되지 않고 끝까지 실행되고 빠져나왔다.

타임아웃

모든 일시중단 함수에는 지정한 시간 이후에 함수를 중단시키는 타임아웃 지정이 필수이다.
withTimeout 함수를 사용해 타임아웃이 설정된 코루틴 Context 생성이 가능하다.

fun main() {
    val job = CoroutineScope(Dispatchers.IO).launch {
        withTimeout(5000) {
            repeat(10) { i ->
                println("repeat: $i")
                delay(1000)
            }
        }
    }
    runBlocking { job.join() }
    println("done")
}
/* 
repeat: 0
repeat: 1
repeat: 2
repeat: 3
repeat: 4
done
*/

설정한 시간 이후에 TimeoutCancellationException 을 발생시킨다.

withTimeout 역시 중단지점이 있어야 발생가능하기 때문에, 위 예제의 중단지점 인 delay 를 Thread.sleep 으로 변경할 경우 타임아웃으로 중단되지 않고 계속 수행된다.

코루틴 예외처리

코루틴은 fire & forgot 디자인을 사용하다보니 실제 코루틴 Scope 의 코드 동작이 어디서 실행될지 모른다.

fun main() {
    runBlocking {
        try {
            // async {
            launch {
                delay(1000)
                throw Error("Some error")
            }
        } catch (e: Exception) {
            println("error class:${e.javaClass.simpleName}, msg:${e.message}")
        }
    }
    println("Done")
}

위와 같이 async, launch 외부에 try catch 정의를 하여도, 코루틴 Scope 정의만 하였을 뿐 실행은 다른곳에서 진행되기 때문에 예외가 걸리지 않는다.
차라리 코루틴 Scope 내부에 try catch 블록을 정의해야 한다.

launch 예외처리

launch 의 경우 예외발생시점을 특정지을 수 없어 자동예외전파(propagating exceptions automatically) 속성을 가지고 있다.

내부에 try catch 블록을 정의하거나 예외핸들러 를 사용해 예외전파를 막아야한다.

CoroutineExceptionHandler 객체를 정의하고 코루틴 Scope 에 적용한다.
코루틴 Context 정보를 함께 전달해 분기처리도 가능하다.

fun main() {
    val exceptionHandler = CoroutineExceptionHandler { context, exception ->
        println("예외처리 $exception, context: ${context[CoroutineName]?.name}")
    }
    // inner try catch block
    val job1 = CoroutineScope(Dispatchers.Default + CoroutineName("MyCoroutine"))
        .launch {
            try {
                throw IllegalArgumentException("예외 테스트1")
            } catch (exception: Exception) {
                println("예외처리 $exception, context: ${coroutineContext[CoroutineName]?.name}")
            }
        }
    // exception handler
    val job2 = CoroutineScope(Dispatchers.Default + CoroutineName("MyCoroutine"))
        .launch(exceptionHandler) {
            throw IllegalArgumentException("예외 테스트2")
        }
    runBlocking {
        job1.join()
        job2.join()
    }
    println("done")
}
/* 
예외처리 java.lang.IllegalArgumentException: 예외 테스트2, context: MyCoroutine
예외처리 java.lang.IllegalArgumentException: 예외 테스트1, context: MyCoroutine
done
*/

예외가 발생하면 기존 실행하던 코루틴 Scope 에서 예외핸들러의 코루틴 Scope 로 실행을 옮긴다.

주의할점은 CancellationException 예외는 예외핸들러 에서 처리하지 않는다.
만약 위 코드에서 IllegalArgumentException 대신 CancellationException 예외를 발생시키면 그냥 코루틴 Scope 를 빠져나온다.

예외핸들러는 코루틴 예외처리를 위한 중복코드를 줄여주지만 안타깝게도 모든 launch 에서 예외핸들러가 동작하진 않는다.
위 예제처럼 launch 가 해당 코루틴 Scope 에 바로 정의된 코루틴 Context 에서만 예외 핸들러가 동작한다.

즉 top level context 에서만 이여야만 핸들러가 동작한다 할 수 있다.

fun main() {
    val h1 = CoroutineExceptionHandler { _, e -> println("handler1 e: $e") }
    val h2 = CoroutineExceptionHandler { _, e -> println("handler2 e: $e") }
    val job = CoroutineScope(Dispatchers.Default)
        .launch(h1) {
            println("start launch1")
            launch(h2) {
                println("start launch2")
                throw RuntimeException("error!")
            }
            println("end launch1")
        }
    runBlocking { job.join() }
}
/* 
start launch1
end launch1
start launch2
handler1 e: java.lang.RuntimeException: error!
*/

위 예에서도 자식 코루틴 Context 에 정의된 h2 예외핸들러 가 사용되지 않고 top level context 에 정의된 h1 예외핸들러 가 호출된다.

만약 top level context 에 예외핸들러 를 정의하지 않았다면 오류가 제어되지 않아 비정상 종료된다.

async 예외처리

async 는 Deferred 객체가 반환되고 await 함수가 해당 코루틴 Scope 의 완료시점임을 알 수 있다.
따라서 코루틴 Scope 내부에서 예외가 발생해도 실제 예외를 던지는 시점을 await 까지 늦추고 await 를 try catch 블록으로 감싸 예외처리를 진행한다.

fun main() {
    val result: Deferred<Int> = CoroutineScope(Dispatchers.Default)
        .async {
            println("코루틴 시작")
            // 예외 발생
            throw IllegalArgumentException("예외 테스트")
            println("코루틴 종료")
            1000; // 반환
        }
    runBlocking {
        try {
            println("result:${result.await()}")
        } catch (exception: Exception) {
            println("예외처리 $exception")
        }
        println("프로그램 종료")
    }
}
/*
코루틴 시작
예외처리 java.lang.IllegalArgumentException: 예외 테스트
프로그램 종료
*/

예외전파, SupervisorJob

코루틴에서 예외가 발생하면 부모 코루틴 까지 전파된다.
예외가 발생하면 코루틴은 취소되고, 이는 부모 코루틴 취소로 이어진다.

처리되지 않은 예외는 아래 그림과 같이 코루틴 Scope 내의 모든 코루틴 취소로 이어지게 된다.

Coroutine#6 에서 발생한 예외로 인해 Coroutine#1 까지 예외가 전파되고 그 아래의 모든 자식 코루틴 Scope 에 Cancels 요청이 전파되었다.

이를 막기위해 예외가 전파되지 않도록 별도의 작업을 수행해햐 한다(예외헨들러, SupervisorJob)

fun main() {
    var outerJob1 = CoroutineScope(Dispatchers.Default).launch {
        println("outerJob1 start")
        var innerJob1 = launch {
            println("innerJob1 start")
            repeat(1000) { i ->
                delay(1000)
                println("innerJob1 $i")
            }
        }
        // 2초 후 error 발생
        var innerJob2 = launch {
            println("innerJob2 start")
            repeat(2) { i ->
                delay(1000)
                println("innerJob2 $i")
            }
            throw Error("error")
        }
        repeat(1000) { i ->
            delay(1000)
            println("outerJob1 $i")
        }
    }
    runBlocking { outerJob1.join() }
}
/* 
outerJob1 start
innerJob1 start
innerJob2 start
outerJob1 0
innerJob2 0
innerJob1 0
outerJob1 1
innerJob1 1
innerJob2 1
Exception in thread "DefaultDispatcher-worker-2 @coroutine#1" java.lang.Error: error
*/

innerJob2 에서 예외가 발생하고 top level context 에게 까지 예외가 전파되면서 예하의 모든 코루틴들도 취소되면서 메인 스레드가 종료되었다.

만약 innerJob2 만 취소시키고 더이상의 예외전파를 원하지 않는다면 SupervisorJob 을 코루틴 Context 에 적용해야 한다.

fun main() {
    var outerJob1 = CoroutineScope(Dispatchers.Default).launch {
        println("outerJob1 start")
        var innerJob1 = launch {
            println("innerJob1 start")
            repeat(1000) { i ->
                delay(1000)
                println("innerJob1 $i")
            }
        }
        // 2초 후 error 발생
        var innerJob2 = launch(SupervisorJob()) {
            println("innerJob2 start")
            repeat(2) { i ->
                delay(1000)
                println("innerJob2 $i")
            }
            throw Error("error")
        }
        repeat(1000) { i ->
            delay(1000)
            println("outerJob1 $i")
        }
    }
    runBlocking { outerJob1.join() }
}

또는 supervisorJobScope 를 통해 SupervisorJob 이 적용된 별도의 코루틴 Scope 생성을 통해 처리한다.

fun main() {
    var outerJob1 = CoroutineScope(Dispatchers.Default).launch {
        println("outerJob1 start")
        var innerJob1 = launch {
            println("innerJob1 start")
            repeat(1000) { i ->
                delay(1000)
                println("innerJob1 $i")
            }
        }
        // 2초 후 error 발생
        supervisorScope {
            var innerJob2 = launch(SupervisorJob()) {
                println("innerJob2 start")
                repeat(2) { i ->
                    delay(1000)
                    println("innerJob2 $i")
                }
                throw Error("error")
            }
            repeat(1000) { i ->
                delay(1000)
                println("outerJob1 $i")
            }
        }
    }
    runBlocking { outerJob1.join() }
}