[Kotlin Coroutine] async를 사용한 동시성. API 호출 결과를 합산

- 해당 문서와 번역본을 참고했습니다.

https://kotlinlang.org/docs/composing-suspending-functions.html#sequential-by-default

Composing suspending functions | Kotlin

https://seyoungcho2.github.io/CoroutinesKoreanTranslation/undefined-1.html

일시중단 함수 구성하기 · GitBook

- 간혹, 서비스를 구성하면서 여러 개의 API를 호출하여 그 결과를 모두 받은 다음에 상대적인 처리가 필요한 경우가 있다.

 

- 예를 들어 1번 2번 3번 API를 호출해서 모두 값이 내려오면 정책적으로 정해진 우선순위에 따라 Response를 처리한다는 등...

 

- API 호출 시 Coroutine을 사용하는데 이를 통해 동시성을 제어하여 모든 Response의 확인을 진행한다.

 

- 일반적인 코드와 같이 Coroutine 코드는 기본적으로 순차적 실행이기 때문에 여러개의 호출 프로세스 뒤, 결과를 전체 합산하거나 검사하기 위해 아래와는 다른 방식의 구성이 필요하다.

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

 

- 그래서 결국 여러 작업 실행 사이에 종속성이 없고, 동시에 처리하고 싶다면 async를 사용해야한다.

 

- 개념적으로 async는 launch와 같다. async는 다른 스레드들과 동시에 동작하는 별도의 경량 Thread인 Coroutine을 시작한다.

 

- 다른 점은 launh는 결괏값을 전달하지 않는 Job을 return 하지만, async의 경우에는 나중에 결과를 반환할 것을 약속하고 Thread Blocking을 하지 않는 Deferred를 반환한다.

 

- 이후에 이 반환된 Deferred에 await() 함수를 호출하여 실질적인 결과 값을 반환받을 수 있으며, Deferred 또한 Job 이기 때문에 취소도 가능하다.

public interface Deferred<out T> : Job {

 

- 그래서 결국 동시에 실행하고 처리가 가능하다. 순차적 실행을 조정할 수 있기에 처리 시간 또한 줄일 수 있다.

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

 

- 또한 즉시 실행이 아니라 개발자가 직접 Coroutine의 실행 제어 권한을 가질 수도 있다.

 

- async의 start 파라미터를 CorountineStart.LAZY로 설정해 주며 된다. 이를 통해 async를 지연적으로 만들어 await에 의해 반환이 필요하거나 Job의 start 함수를 통해 직접 실행시킬 수 있다.

val time = measureTimeMillis {
    val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
    val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
    // some computation
    one.start() // 첫 째를 start
    two.start() // 둘 째를 start
    println("The answer is ${one.await() + two.await()}")
}
println("Completed in $time ms")

 

- 구조적인 동시성에서 벗어나기 위해 GlobalScope를 참조하는 async Coroutine 빌더를 사용하여 각 api 호출 로직을 실행하는 비동기 스타일의 함수 구성이 가능하다.

// somethingUsefulOneAsync 의 반환 타입은 Deferred<Int> 이다.
@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulOneAsync() = GlobalScope.async {
    doSomethingUsefulOne()
}

// somethingUsefulTwoAsync 의 반환 타입은 Deferred<Int>
@OptIn(DelicateCoroutinesApi::class)
fun somethingUsefulTwoAsync() = GlobalScope.async {
    doSomethingUsefulTwo()
}

- 위 비동기 스타일 함수들은 모두 suspend 함수가 아니다.

 

- 아래는 위 함수들이 Corountine 영역 외부에서 사용되는 모습이다.

fun main() {
    val time = measureTimeMillis {
        // 우리는 Coroutine 바깥에서 비동기 작업을 시작 할 수 있다. 
        val one = somethingUsefulOneAsync()
        val two = somethingUsefulTwoAsync()
        // 하지만 결과를 기다리는 것은 일시중단이나 블로킹 중 하나를 포함해야 한다.
        // 여기서 우리는 `runBlocking { ... }` 을 사용해 메인 스레드를 블록시키고 결과값이 오기를 기다린다.
        runBlocking {
            println("The answer is ${one.await() + two.await()}")
        }
    }
    println("Completed in $time ms")
}

- 여기서 만일 val one = somethingUsefulOneAsync() 행과 one.await() 표현식 사이에 약간의 논리 오류가 발생해, 프로그램이 예외를 발생시켜 프로그램에 의해 수행되던 작업이 중단되면 전역 오류 처리기는 해당 예외를 잡아 보고를 할 수 있겠지만, 작업이 중단되었음에도 비동기 동작하는 somethingUsefulOneAsync()는 계속 실행 중일 것이다.

 

- 이를 위해 구조화된 동시성 적용이 필요하다.

 

- async를 사용한 동시 실행 예제를 사용하여 doSomethingUsefulOne과 doSomethingUsefulTwo를 동시에 실행하고 그들의 실행 결과를 합쳐서 반환하는 함수를 추출해 보자

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

- async Coroutine Builder가 CoroutineScope의 확장 함수로 정의되어 있기 때문에 이를 Scope내에 포함해야 하며, 이것이 coroutineScope 함수가 제공하는 기능이다.

 

- 이렇게 하면 concurrentSum 내부에서 예외 발생 시, Scope 내부에서 실행된 모든 Coroutine이 취소된다.

 

- 취소는 언제나 Coroutines의 계층 구조를 통해 전파된다.

import kotlinx.coroutines.*

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) // Emulates very long computation
            42
        } finally {
            println("First child was cancelled")
        }
    }
    val two = async<Int> { 
        println("Second child throws an exception")
        throw ArithmeticException()
    }
    one.await() + two.await()
}

- 자식들 중 하나(위에서는 two라는 변수로 명명됨)가 취소로 인해 실패하면 첫 async 함수와 await을 수행 중인 부모가 모두 취소되는 방식에 유의하자

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

 

- 결론적으로 API를 동시 호출하여 값을 일시적으로 처리하기 위해 아래와 같이 다중 API 호출 및 처리를 구성할 수 있다.

private fun fetchData() = CoroutineScope(Dispatchers.IO).launch {
    val first = async { try {
        firstApiCall()
    } catch (e: Exception) { null } }

    val second = async { try {
        secondApiCall()
    } catch (e: Exception) { null } }

    val third = async { try {
        thirdApiCall()
    } catch (e: Exception) { null } }

    withContext(Dispatchers.Main) {
        val firstData = first.await()
        val secondData = second.await()
        val thirdData = third.await()
        .
        .
        .