[Kotlin] Coroutines - Suspending 함수 활용하기

다양한 suspending 함수를 조합해 보자.

순차적으로 실행된다

다음의 두 suspending 함수가 있다. 두 함수는 어떤 유용한 계산을 수행한 후 결과를 반환한다.

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // 유용한 척
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // 유용한 척..
    return 29
}

두 함수를 순서대로 부르면 어떻게 될까? 순서대로 부른다는 것은 작업의 순서가 있다는 의미가 있다. 예를 들어 첫 번째 함수의 결과값에 따라 두 번째 함수를 부를 수도, 부르지 않을 수도 있지 않겠는가?

val time = measureTimeMillis {
    val one = doSomethingUsefulOne()
    val two = doSomethingUsefulTwo()
    println("The answer is ${one + two}")
}
println("Completed in $time ms")

코루틴 내부의 코드는 보통의 코드처럼 순서대로 실행된다. 위 코드의 수행 시간을 보면 알 수 있다.

The answer is 42
Completed in 2017 ms

동시에 해 보자 (concurrent)

사실 위의 두 함수는 서로 연관관계가 없다. 이처럼 여러 코드를 동시에 수행해도 될 때는 async를 사용하면 좋다. async는 launch와 비슷한 코루틴 빌더이지만, Job 대신 Deferred 객체를 반환한다는 차이점이 있다. Deferred는 결과값을 나중에 돌려준다는 약속의 의미가 있다. .await() 함수를 사용하면 async가 완료됐을 때 결과값을 반환받을 수 있다. 물론 Deferred도 중간에 취소할 수 있다.

val time = measureTimeMillis {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")

The answer is 42
Completed in 1017 ms

두 함수를 거의 동시에 실행하여 실행 시간을 대폭 단축하였다. 코루틴에서 코드를 동시에 실행하려면 항상 이렇게 명시적으로 코드를 작성해야 한다(async 등).

나중에 실행하고 싶으면?

async로 만들어진 코루틴을 즉시 시작하지 않을 수도 있다. start = CoroutineStart.LAZY 옵션을 주면 코루틴이 즉시 실행되지 않으며, 나중에 .start() 함수를 호출하여 시작할 수 있다.

val time = measureTimeMillis {
    val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
    val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
    // ...
    one.start() // one 시작
    two.start() // two 시작
    println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")

The answer is 42
Completed in 1017 ms

Deferred는 Job을 상속받았기 때문에 Deferred로도 코루틴의 실행을 제어할 수 있다. 참고로 await 함수는 코루틴이 아직 시작되지 않았을 경우 코루틴을 시작한다.

계산 과정에서 suspending 함수가 사용되는 경우 코틀린의 lazy 대신 CoroutineStart.LAZY 옵션을 사용할 수 있다.

비동기! (asynchronous ≒ concurrent)

GlobalScope 범위에서 async 빌더를 이용하여 비동기적으로 작동하는 함수를 정의할 수 있다. 함수 이름 뒤에 ...Async를 붙여서 함수가 비동기적으로 작동한다는 점을 명시하도록 하자.

// Deferred<Int> 반환
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

// Deferred<Int> 반환
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

주의: xxxAsync 함수는 suspending이 아니므로 어디에서든 사용될 수 있지만, 실제 동작은 비동기적으로 이루어진다. 다음의 코드를 보자.

// runBlocking으로 시작하지 않는다!
fun main() {
    val time = measureTimeMillis {
        // 두 코루틴은 비동기적으로 실행되므로 스레드를 block하지 않는다.
        val one = somethingUsefulOneAsync()
        val two = somethingUsefulTwoAsync()
        // 하지만 결과값을 기다리려면 suspend 또는 block되어야 한다.
        // 여기서는 runBlocking을 사용하여 결과값을 기다리는 동안 메인 스레드를 block하였다.
        runBlocking {
            println("The answer is ${one.await() + two.await()}")
        }
    }
    println("Completed in $time ms")
}

The answer is 42
Completed in 1122 ms

참고로 코틀린에서는 이런 코드를 매우 강력하게 권장하지 않는다(strongly discouraged). val one = somethingUsefulOneAsync()와 one.await() 사이에서 Exception이 발생하여 실행이 중단되었다고 가정해 보자. 일반적인 경우 try catch 등을 이용하여 에러를 처리하므로 문제가 없다. 그런데 비동기적으로 실행된 one과 two는 자신을 호출한 부분이 중단되었음에도 불구하고 계속해서 계산을 수행하고 있다.

이런 문제를 해결하려면 structured concurrency 개념을 사용해야 한다.

async와 structured concurrency

위에 있는 뭔가 유용해 보이는 두 함수를 실행하여 결과값의 합을 반환하는 함수를 만들어 보자. async 빌더는 CoroutineScope의 확장 함수이기 때문에 특정 scope 안에서 async를 실행해야 한다. 바로 그럴 때 coroutineScope 함수를 이용한다.

suspend fun concurrentSum(): Int = coroutineScope {
    val one = async { doSomethingUsefulOne() }
    val two = async { doSomethingUsefulTwo() }
    one.await() + two.await()
}

이렇게 하면 concurrentSum 안에서 Exception이 발생했을 때 concurrentSum 안의 모든 코루틴이 종료된다. one과 two는 여전히 비동기적으로 실행된다.

val time = measureTimeMillis {
    println("The answer is ${concurrentSum()}")
}
println("Completed in $time ms")

The answer is 42
Completed in 1028 ms

자식 코루틴이 취소되면 부모 코루틴에도 취소가 전달된다.

fun main() = runBlocking<Unit> {
    try {
        failedConcurrentSum()
    } catch(e: ArithmeticException) {
        println("Computation failed with ArithmeticException")
    }
}

suspend fun failedConcurrentSum(): Int = coroutineScope {
    val one = async<Int> { 
        try {
            delay(Long.MAX_VALUE) // 매우 오랫동안 계산하는 척
            42
        } finally {
            println("First child was cancelled")
        }
    }
    val two = async<Int> { 
        println("Second child throws an exception")
        throw ArithmeticException() // 코루틴이 취소된다.
    }
    one.await() + two.await()
}

Second child throws an exception
First child was cancelled
Computation failed with ArithmeticException

failedConcurrentSum에서 발생한 Exception이 main의 runBlocking까지 전달된 것을 확인할 수 있다. 이처럼 코루틴의 계층 구조를 통해 코루틴을 제어할 수 있다.

참고 문헌

Coroutine Composing Suspending Function - kotlinlang