[Kotlin] 코틀린-코루틴#3 - suspending function의 구성

이 글은 아래 링크의 내용을 기반으로 하여 설명합니다.

https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md

또한 예제에서 로그 print시 println과 안드로이드의 Log.e()를 혼용합니다.

Sequential is default

기본적으로 code는 sequential하게 수행됩니다. 이는 coroutine 내부에서도 동일하게 적용 됩니다. (당연하겠죠?)

먼저 아래와 같이 시간이 걸리는 작업을 하는 의미있는 함수가 있다고 가정합니다.

(remote service call이 될수도 있고, computation을 하는 작업일수도 있습니다.)

먼저, 예제에서 공통으로 사용하기 위한 suspend 함수를 생성합니다.

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here, too
    return 29
}

두개의 작업을 하는데는 얼마간의 시간이 필요하므로 여기서는 delay(1000L)으로 대신하여 표현하겠습니다.

fun main() = runBlocking<Unit> {
    val time = measureTimeMillis {
        val one = doSomethingUsefulOne()
        val two = doSomethingUsefulTwo()
        println("The answer is ${one + two}")
    }
    println("Completed in $time ms")
}

위 예제에서는 doSomtingUsefullOne()과 doSomtingUsefullTwo()를 순차적으로 호출하며, 이때 나온 값을 더하여 출력합니다.

결과값은 아래와 같습니다.

The answer is 42

Completed in 2017 ms

두개의 덧셈을 수행하는데 2000ms이 조금 더 걸렸습니다.

한개의 함수가 1000ms씩 잡아먹으니 당연한 결과 입니다.

Concurrent using async

위 예제의 두개의 함수는 서로 인과관계를 갖지 않습니다.

따라서 따로 동시에 동작하면 더 빠른 값을 얻을수 있겠지요?

이럴때 coroutine 내부에서 async를 사용하여 두개의 작업을 분리 및 동시진행 시킬 수 있습니다.

async과 launch는 사실 동일한 컨셉입니다만 return하는 객체가 다릅니다.

• launch -> Job return
• async -> Deferred return

Job은 launch 시킨 코루틴의 상태를 관리하는 용도로 사용되는 반면 코루틴의 결과를 리턴받을 수 없으나, Deferred는 async 블럭에서 수행된 결과를 return 받을 수 있습니다.

물론 Deferred는 Job을 상속받아 구현되었기 때문에 상태제어 또한 가능합니다.

fun main() = runBlocking {
    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는 자바로 표현하자면 CompletableFuture.supplyAsync()와 같습니다.

또한 Deferred역시 Future와 같은 기능이라고 볼수 있습니다.

Lazily started sync

async로 시작되는 coroutine은 블랙내 코드의 수행을 지연시킬 수 있습니다.

optional 파라미터인 start의 값을 CoroutineStart.LAZY 로 주면 해당 작업을 바로 수행하지 않습니다.

작업을 수행하려면 start()를 명시적으로 불러주거나 await()를 호출하면 됩니다.

fun main() = runBlocking {
    val time = measureTimeMillis {
        val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
        val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
        // some computation
        one.start() // start the first one
        two.start() // start the second one
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
}

위 코드에서는 명시적으로 one.start(), two.start()를 호출하여 async를 수행합니다.

따라서 동시에 두개가 parallel하게 수행되어 아래와 같이 로그가 찍힙니다.

The answer is 42

Completed in 1017 ms

단 start()를 삭제하더라도 println 내부의 await()를 만나면서 해당 블럭은 실행 됩니다.

단 await()는 실행이 완료될때까지 대기하므로. one.await()의 작업이 완료되어야 two.await()가 동작합니다. (sequential 하게 동작함)

await()만 사용하는 경우 이렇게 의도하지 않게 동작하지 않도록 주의해야 합니다.

만약 자바로 구현한다면, stream을 만들고 stream 안에서 completableFuture.supplyAsync를 구현하면 가능할것 같네요.

Async-style functions

GlobalScoped에서 async coroutine builder를 사용하여 명시적으로 비동기 형식의 함수를 만들수 있습니다.

// The result type of somethingUsefulOneAsync is Deferred
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

// The result type of somethingUsefulTwoAsync is Deferred
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

위 함수는 GlobalScope을 이용해서 명시적으로 async 함수를 만들었고, suspend function은 아닙니다.

따라서 어디서든 호출되어 사용할 수 있습니다.

// note, that we don't have `runBlocking` to the right of `main` in this example
fun main() {
    val time = measureTimeMillis {
        // we can initiate async actions outside of a coroutine
        val one = somethingUsefulOneAsync()
        val two = somethingUsefulTwoAsync()
        // but waiting for a result must involve either suspending or blocking.
        // here we use `runBlocking { ... }` to block the main thread while waiting for the result
        runBlocking {
            println("The answer is ${one.await() + two.await()}")
        }
    }
    println("Completed in $time ms")
}

실제로 값을 얻기위해서(block 하고 대기하기 위해서) await()는 runBlocking{..}에서 받아 옵니다.

만약 val one = somthingUsefullOneAsync() 와 one.await()사이에서 exception이 발생하면 프로그램이 종료됩니다. 

외부의 try-catch로 처리하여 exception handling은 할 수있으나 비동기 작업이 유지된채로 남습니다.

따라서 이런식의 비동기 함수 사용은 다른 프로그램에서도 많이 사용하는 패턴이지만 kotlin coroutine에서는 강력하게 비추천 합니다.

참고로 measureTimeMillis{..}는 coroutine api가 아니라 kotlin 기본 api 입니다.

따라서 아래와 같이 함수를 만들어 사용하면 좀더 메모리 leak에 안전한 프로그램을 만들 수 있습니다.

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

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

결과

The answer is 42

Completed in 1017 ms

async coroutine builder는 CotourineScope의 extention function 입니다.

따라서 위처럼 부모 Scope을 새로 만들고 그안에서 async를 돌린다면, 두개의 aysnc는 같은 부모 scope에서 수행된 작업들이 됩니다.

따라서 scope 안에서 exception이 발생하면 해당 scope을 빠져나가면서 해당 coroutineScope에서 수행되었던 자식 coroutine들도 다 취소 됩니다.

 fun scopeException() {
        runBlocking {
            try {
                failedConcurrentSum()
            } catch (e: ArithmeticException) {
                Log.e(TAG, "Computation failed with ArithmeticException")
            } finally {
                Log.e(TAG, "Main fianlly")
        }
    }

    suspend fun failedConcurrentSum(): Int = coroutineScope {
        val one = async {
            try {
                delay(Long.MAX_VALUE) // Emulates very long computation
                42
            } finally {
                Log.e(TAG,"First child was cancelled")
            }
        }

        val two = async {
            Log.e(TAG,"second async")
            try {
                delay(Long.MAX_VALUE) // Emulates very long computation
                42
            } catch(e: CancellationException) {
                Log.e(TAG,"second child was cancelled")
            }
        }

        val three = async {
            Log.e(TAG,"Third child throws an exception")
            throw ArithmeticException()
        }

        one.await() + two.await() + three.await()
    }

결과

second async

Third child throws an exception

First child was cancelled

second child was cancelled

Computation failed with ArithmeticException

Main finally

만약 중간에 GlobalScope.launch{..}를 넣었다면 이 동작은 취소되지 않습니다.

(여기서 launch는 GlobalScope에서 시작된 작업이므로 coroutineScope{..}으로 만든 scope의 자식이 아님)

이를 분리시키거나 연결시키기 위한 내용은 다음글에서 다룹니다.