[Coroutine] async와 Deferred

개요

코틀린 코루틴의 정석

 

오늘은 코루틴의 async와 Deferred에 대해 정리해보려고 한다.

 

1. async와 Deferred

async는 launch와 마찬가지로 새로운 코루틴을 실행하는 빌더 함수이다.

다만, 차이점이 있다면 launch에 경우 결괏값을 직접 반환할 수 없고, async는 Deferred<T> 타입의 객체를 반환한다.

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

 

여기서 Deferred는 Job과 같이 코루틴을 추상화한 객체로 코루틴으로부터 생서된 결괏값을 감싸는 기능을 가진다.

 

Deferred의 제네릭 타입을 지정하기 위해 명시적으로 타입을 설정해도 되고, async 블록의 반환값으로 반환할 결괏값을 설정해도 된다.

 

fun main() = runBlocking {
    val deferred: Deferred<Int> = async(Dispatchers.IO) {
        delay(1000)
        return@async 42
    }
}

 

이 코드에서 async 코루틴 빌더는 Dispatchers.IO를 사용했고, 1초 지연 후 42를 결과로 반환하는 코루틴을 만든다.

이때 반환 타입은 Deferred<Int>가 된다.

 

2. await를 통한 결괏값 수신

Deferred 객체는 결괏값의 수신의 대기를 위해 await 함수를 제공한다. 마치 lanch에서 join을 쓰는 느낌?

 

await의 대상이 된 Deferred 코루틴이 실행 완료될 때까지 await 함수를 호출한 코루틴을 일시 중단하고, 실행 완료되면 결괏값을 반환하고 호출한 코루틴을 재개한다.

 

fun main() = runBlocking {
    val deferred = async(Dispatchers.IO) {
        delay(1000)
        "Hello, World!"
    }
    println("Waiting...")
    val result = deferred.await()
    println("Result: $result")
}

 

deferred.await()를 호출하면 코루틴 실행이 완료될 때까지 runBlocking 코루틴이 일시 중단된다.

이후에 Hello,World!가 반환되면 runBlocking 코루틴이 재개되고, result 변수에 할당된다.

 

3. Deferred와 Job

Deferred 객체는 Job 객체의 특수한 형태로 Deferred 인터페이스가 Job 인터페이스의 서브타입으로 선언된 인터페이스이다.

즉, Deferred 객체는 코루틴으로부터 결괏값 수신을 위해 Job 객체에서 몇 가지 기능이 추가된 것 뿐, Job 객체의 일종이다.

public interface Deferred<out T> : Job {

    public suspend fun await(): T

    public val onAwait: SelectClause1<T>

    @ExperimentalCoroutinesApi
    public fun getCompleted(): T

    @ExperimentalCoroutinesApi
    public fun getCompletionExceptionOrNull(): Throwable?
}

 

Job 객체의 일종이기 때문에, Job 객체의 모든 함수 및 프로퍼티를 사용할 수 있다.

join을 사용할 수 있고, cancel 함수를 호출해 취소도 할 수 있다.

또한 상태 조회를 위한 isActive, isCancelled, isCompleted 같은 프로퍼티들을 사용할 수 있다.

 

4. 복수의 코루틴 적용

4-1. 순차 처리

fun main() = runBlocking {
    val deferred1 = async { task1() }

    val result1 = deferred1.await()

    val deferred2 = async { task2() }
    val result2 = deferred2.await()

    println("Sum: ${result1 + result2}")
}

suspend fun task1(): Int {
    delay(1000)
    println("Task 1 completed")
    return 10
}

suspend fun task2(): Int {
    delay(2000)
    println("Task 2 completed")
    return 20
}

 

코드의 흐름을 정리해보자면, 다음과 같다.

 

1) deferred1 코루틴을 통해 task1 작업이 실행된다.

2) task1 작업이 실행 완료될 때까지 runBlocking 코루틴이 일시 중지되었다가, 실행 완료되면 result1 변수에 값이 할당된다.

3) deferred2 코루틴을 통해 task2 작업이 실행된다.

4) 2번과 마찬가지로 중지되었다가, task2 작업이 완료되면 result2 변수에 값이 할당된다.

5) Sum: 30이 출력된다.

 

즉, await()을 호출하는 시점에 runBlocking 코루틴이 일시 중지되기 때문에 순차적으로 처리되게 된다.

 

하지만 서로 독립적인 작업인 deferred1과 deferred2를 순차적으로 처리한다면 상당히 비효율적이다. 그럼 이를 동시에 처리할 수 있는 방법은 있을까?

4-2. 병렬 처리

fun main() = runBlocking {
    val deferred1 = async { task1() }
    val deferred2 = async { task2() }
    
    val result1 = deferred1.await()
    val result2 = deferred2.await()

    println("Sum: ${result1 + result2}")
}

suspend fun task1(): Int {
    delay(1000)
    println("Task 1 completed")
    return 10
}

suspend fun task2(): Int {
    delay(2000)
    println("Task 2 completed")
    return 20
}

 

답은 간단하다.

deferred1.await()이 호출되기 전에 deferred2가 실행되도록 순서를 바꿔주면 된다.

 

4-3. awaitAll

위에서의 코드는 task가 2개여서 이렇게 작성해도 괜찮았는데 만약 10개 이상이 된다면??? 일일이 다 await()을 호출해서 처리하기에는 비효율적이다.

 

이때 사용할 수 있는 것이 awaitAll 함수이다.

public suspend fun <T> awaitAll(vararg deferreds: Deferred<T>): List<T>

 

awaitAll 함수는 가변 인자로 Deferred 타입의 객체를 받아 인자로 받은 모든 Deferred 코루틴으로부터 결과가 수신될 때까지 호출부의 코루틴을 일시 중단하고, 결괏값들을 List로 만들어 반환한다.

 

fun main() = runBlocking {
    val deferred1 = async { task1() }
    val deferred2 = async { task2() }

    val result: List<Int> = awaitAll(deferred1, deferred2)
    println("Sum: ${result[0] + result[1]}")
}

suspend fun task1(): Int {
    delay(1000)
    println("Task 1 completed")
    return 10
}

suspend fun task2(): Int {
    delay(2000)
    println("Task 2 completed")
    return 20
}

 

컬렉션에 사용

awaitAll 함수를 Collection 인터페이스에 대한 확장 함수로도 제공한다.

public suspend fun <T> Collection<Deferred<T>>.awaitAll(): List<T>

 

fun main() = runBlocking {
    val time = measureTimeMillis {
        val deferredList = listOf(
            async { fetchData(1) },
            async { fetchData(2) },
            async { fetchData(3) }
        )

        // 모든 비동기 작업이 완료될 때까지 대기하고 결과를 리스트로 받음
        val results = deferredList.awaitAll()

        println("Results: $results")
    }

    println("Completed in $time ms")
}

suspend fun fetchData(id: Int): String {
    delay(1000L)
    return "Result from $id"
}

// Results: [Result from 1, Result from 2, Result from 3]
// Completed in 1005 ms

 

5. withContext

withContext 함수를 사용한다면 async-await 작업을 대체할 수 있다.

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

 

withContext 함수가 호출되면 CoroutineContext 객체를 사용해 block 람다식을 실행하고, 완료된다면 결과를 반환한다.

 

이것을 활용해서 async-await을 대신해보면

 

fun main() = runBlocking {
    val result = withContext(Dispatchers.IO){
        delay(1000L)
        return@withContext 42
    }
    println(result)
    
    //42
}

 

Deferred 객체를 생성하는 과정 없이 즉시 42라는 결과가 반환되도록 작성할 수 있다.

5-1. 동작 방식

withContext 함수는 새로운 코루틴을 만드는 것이 아니라 실행 중이던 코루틴을 그대로 유지한 채로 코루틴의 실행 환경만 변경해서 작업을 처리한다.

 

fun main() = runBlocking {
    println("[${Thread.currentThread().name}] runBlocking ")
    withContext(Dispatchers.IO){
        println("[${Thread.currentThread().name}] withContext ")
    }
}

 

withContext에서의 CoroutineContext 객체가 Dispatchers.IO로 바뀌었기 때문에 백그라운드 스레드(= DefaultDispatcher-worker-1)에서 실행됐다.

 

 

withContext 함수가 호출되면 실행 중인 코루틴의 실행 환경이 withContext 함수의 context 인자 값으로 변경되어 실행된다. =(Context Switching)

 

즉, 위의 그림처럼 withContext(Dispatchers.IO)가 호출되면 해당 코루틴은 다시 Dispatchers.IO의 작업 대기열로 이동한 후 Dispatchers.IO가 사용할 수 있는 스레드 중 하나로 보내져서 실행된다.

 

5-2. 주의점

withContext 함수는 위에서 언급한 것 처럼 새로운 코루틴을 만들지 않기 때문에 하나의 코루틴에서 withContext 함수가 여러번 호출 된다면 순차적으로 실행되게 된다.

 

즉, 병렬로 작업해야 하는 상황에서 문제가 발생할 수 있다는 것이다.

 

fun main() = runBlocking {
    val result1 = withContext(Dispatchers.IO){
        delay(1000)
        return@withContext "complete task1"
    }
    val result2 = withContext(Dispatchers.IO){
        return@withContext "complete task2"
    }

    println(result1)
    println(result2)

}

 

 

이 경우에 result1이 실행 완료될 때까지 runBlocking 코루틴이 일시 정지되고, result1이 실행 완료된 후에 result2가 실행되게 된다.

 

즉, 이렇게 병렬 처리가 필요한 상황에서는 withContext가 아닌 async-await을 사용해야 한다.