[Kotlin] Coroutines - Coroutine builder

#1 Coroutine builder

kotlinx-coroutines-core

runBlocking을 제외한 코루틴 빌더는 CoroutinesScope의 확장 함수(Extension functions)다. CoroutineScope가 코루틴 코드 영역(Scope)과 아닌 영역을 구분하는 역할이라면, Coroutine builder는 그 CoroutineScope 영역의 코루틴 코드를 수행하는 역할이다. Coroutine builder는 새로운 스레드를 스레드 풀로부터 할당받는 함수이기도 하다. Coroutine builder의 종류는 아래의 4가지다.

#2 kotlinx.coroutines.runBlocking

[Kotlin] Coroutines - runBlocking

runBlocking은 위 게시글에서 다룬다.

 

#3 Coroutine builder - CoroutineScope.launch

#3-1 코드

import kotlinx.coroutines.*

fun performTaskInCoroutineScope(): Job {
    return CoroutineScope(Dispatchers.Default).launch {
        var x = 1
        repeat(5) {
            println(x++)
            delay(1000) // 1초 대기
        }
    }
}

fun main() {
    println("Start")

    val myJob = performTaskInCoroutineScope()

    println("Continuing...")

    runBlocking {
        // 해당 Job의 완료를 기다림
        myJob.join()
    }

    println("End")
}

/* 출력 결과
Start
Continuing...
1
2
3
4
5
End
*/

CoroutineScope.launch는 현재 스레드를 Blocking(블로킹)하지 않고, CoroutineContext에 명시된 스레드에서 코루틴을 실행한다. '현재 스레드를 블로킹한다'는 것은 '해당 스레드의 코드 동작을 일시 정지함'을 의미한다. 블로킹은 현재 스레드와 코루틴용 스레드의 병렬적 동시 사용이라는, 코루틴의 일반적인 목적에 부합하지 않는다. 따라서 가장 일반적이며 대표적인 코루틴 함수인 launch는 현재 스레드를 블로킹하지 않아야 하는 것이다.
 

#3-2 Job 인터페이스

CoroutineScope.launch는 코루틴 코드의 레퍼런스 역할을 하는 Job 인터페이스의 객체를 return한다. 이 레퍼런스 객체를 변수에 할당해 해당 코루틴을 참조(이 링크의 #6-2)할 수도 있다. CoroutineScope.launch는 Job()만을 return한다. 따라서 이 함수에서는, 코루틴 코드 속 어떤 계산의 결과를 가져오거나 할 수 없다.
 

#3-3 Job.join()

join의 사전적 의미는 '연결하다, 합쳐지다'다. join()을 통해, 언제 완료될 지 모르는 어떤 코루틴을 완료된 상태로 현재와 연결짓는다. 말이 어려운데 그냥 Job이 완료될 때까지 기다리는 함수다. 위 코드에서 join()이 runBlocking { ... } 안에 들어가 있는 이유는 여기에 있다.
 

#4 Coroutine builder - CoroutineScope.async

#4-1 코드

import kotlinx.coroutines.*

fun calculateSumAsync(a: Int, b: Int): Deferred<Int> {
    return CoroutineScope(Dispatchers.Default).async {
        println("Calculating sum in background thread...")
        delay(1000) // 1초 동안 대기 (실제로는 어떤 계산을 수행하는 것으로 대체될 수 있음)
        return@async a + b // 계산 결과 반환
    }
}

fun main() {
    println("Start")

    val deferredResult: Deferred<Int> = calculateSumAsync(5, 10) // Deferred 객체 생성

    println("Doing some other work...")

    // 결과를 기다림
    // await() 함수를 호출하면, 비동기 작업이 완료될 때까지 대기한 후 작업 결과를 반환해줌
    runBlocking {
        val result = deferredResult.await()
        println("Result: $result")
    }

    println("End")
}

/* 출력 결과
Start
Doing some other work...
Calculating sum in background thread...
Result: 15
End
*/

CoroutineScope.launch와 마찬가지로 현재 스레드를 Blocking하지 않고, CoroutineContext에 명시된 스레드에서 코루틴을 실행한다. 하지만, CoroutineScope.async은 CoroutineScope.launch와 달리 Job()이 아닌 Deferred<T> 인터페이스의 인스턴스를 return한다.
 

#4-2 Deferred<T> 인터페이스

Deferred<T> 인터페이스는 Job 인터페이스의 자식이다. 따라서 모든 Deferred는 Job이지만, 모든 Job이 Deferred인 것은 아니다. Deferred는 Job을 다뤘던 것처럼 동일하게 참조해 사용할 수 있다. 하지만 Deferred는 Job과 달리, (데이터 형식이 T인) 객체를 저장한다. 그래서 CoroutineScope.async에서는 코루틴 코드 속 어떤 계산의 결과를 가져오거나 할 수 있다.
 
Defer의 사전적 의미는 '미루다, 연기하다'다. Deferred는 수동태니까, '미뤄진, 연기된' 정도의 뜻일테다. 말 그대로 Deferred<T> 인터페이스는 '미뤄진, 연기된' 값을 저장한다. 그도 그럴게, CoroutineScope.async가 수행된 스레드에서 계산이 완료되어야만 비로소 Deferred<T> 객체가 있는 스레드로 결괏값이 전달될 것이기에 필연적으로 '값이 미뤄질 수밖에' 없는 것이다.
 

#4-3 Deferred.await()

'미뤄진' 값을 받는 방식이므로 Deferred가 저장하 값을 받는 멤버 메소드 이름도 '기다리다'라는 뜻의 await다. 위 코드에서 await()가 runBlocking { ... } 안에 들어가 있는 이유는 여기에 있다.
 

#5 Coroutine builder - CoroutineScope.produce

#5-1 코드

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun produceNumbers(): ReceiveChannel<Int> {
    return CoroutineScope(Dispatchers.Default).produce {
        var x = 1
        while (true) {
            send(x++) // 값을 채널로 보냄
            delay(1000) // 1초 대기
        }
    }
}

fun main() {
    val numbers: ReceiveChannel<Int> = produceNumbers() // ReceiveChannel 생성

    repeat(5) {
        runBlocking { println(numbers.receive()) } // 채널에서 값 수신
    }

    println("Done receiving")

    // ReceiveChannel을 닫음
    numbers.cancel()
}

/* 출력 결과
1
2
3
4
5
Done receiving
*/

produce는 launch 및 async와 마찬가지로 현재 스레드를 Blocking하지 않고, CoroutineContext에 명시된 스레드에서 코루틴을 실행한다. produce는 async를 심화시킨 형태다. async은 한 번 값을 return하고 끝나는 데 반해, produce는 계속해서 return 값을 생산(produce)해낸다. 따라서 복수 개의 값을 얻을 수 있다. CoroutineScope.produce는 CoroutineScope.async과 달리 Deferred가 아닌 ReceiveChannel<E> 인터페이스의 인스턴스를 return한다.
 

#5-2 ReceiveChannel<E> 인터페이스

ReceiveChannal은 복수 개의 값을 제공하는 Deferred 비스무리한 것이라는 점에서, Deferred와 마찬가지로 부모가 Job이지 않을까 추측되지만 실제 그렇진 않다. 그래도 느낌으론 비슷하다. Job.cancel과 거의 같은 기능을 하는 ReceiveChannel.cancel 메소드도 지니고 있다.
 

#5-3 ReceiveChannel.receive()

receive()는 받을 때마다 값이 달라진다. 위 코드에서 receive()가 runBlocking { ... } 안에 들어가 있는 이유는 여기에 있다.

#6 CoroutineScope 생략 (Builder 속 Builder 형태 만들기)

#6-1 CoroutineScope 중첩

import kotlinx.coroutines.*

suspend fun main() {
    val myJob = CoroutineScope(Dispatchers.Default).launch {
        println("I'm in launch() function on ${Thread.currentThread().name} thread")
        println("Async will be started after 2 seconds")
        delay(2000)

        val asyncResult = CoroutineScope(Dispatchers.Default).async {
            println("I'm in async() function on ${Thread.currentThread().name} thread")
            println("Calculating...")
            delay(3000)
            return@async 777
        }

        println("asyncResult is ${asyncResult.await()}")
    }

    myJob.join()
}

/* 출력 결과
I'm in launch() function on DefaultDispatcher-worker-1 thread
Async will be started after 2 seconds
I'm in async() function on DefaultDispatcher-worker-2 thread
Calculating...
asyncResult is 777
*/

위 코드처럼 코루틴을 중첩해서 사용하는 경우가 있다. 그러나, 이와 같이 CoroutineScope(CoroutineContext)를 계속 사용하는 것은 코드의 가독성을 상당히 해친다. fun 앞에 있는 suspend 키워드는 일단 본 게시글에선 무시하고, CoroutineScope의 중첩에만 집중하자.

 

#6-2 CoroutineScope 생략 - CoroutineContext 암시적 상속

import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val myJob = CoroutineScope(Dispatchers.Default).launch {
            println("I'm in launch() function on ${Thread.currentThread().name} thread")
            println("Async will be started after 2 seconds")
            delay(2000)

            val asyncResult = async {
                println("I'm in async() function on ${Thread.currentThread().name} thread")
                println("Calculating...")
                delay(3000)
                return@async 777
            }

            println("asyncResult is ${asyncResult.await()}")
        }

        myJob.join()
    }
}

/* 출력 결과
I'm in launch() function on DefaultDispatcher-worker-1 thread
Async will be started after 2 seconds
I'm in async() function on DefaultDispatcher-worker-2 thread
Calculating...
asyncResult is 777
*/

#6-1과 동일한 동작을 하는 코드다. 코틀린에서는 중첩된 Builder의 CoroutineScope(CoroutineContext)를 생략할 수 있다.  코드에서 async()의 CoroutineScope는 launch()와 동일하다.

 

#6-3 CoroutineScope 생략 - CoroutineContext 변경하기

suspend fun main() {
    val myJob = CoroutineScope(Dispatchers.Default).launch {
        println("I'm in launch() function on ${Thread.currentThread().name} thread")
        println("Async will be started after 2 seconds")
        delay(2000)

        val asyncResult = async(Dispatchers.IO + CoroutineName("myAsync")) {
            println("I'm in async() function on ${Thread.currentThread().name} thread")
            println("Calculating...")
            delay(3000)
            return@async 777
        }

        println("asyncResult is ${asyncResult.await()}")
    }

    myJob.join()
}

/* 출력 결과
I'm in launch() function on DefaultDispatcher-worker-1 thread
Async will be started after 2 seconds
I'm in async() function on DefaultDispatcher-worker-2 thread
Calculating...
asyncResult is 777
*/

CoroutineScope는 생략하면서, CoroutineContext만 변경할 수도 있다. 바로, Builder의 매개변수로 CoroutineContext를 전달하면 된다. 문법은 CoroutineScope에 CoroutineContext를 전달하는 것과 동일하다. Dispatchers.Default와 Dispatchers.IO의 스레드명이 같은 이유는 이 링크의 #3-2에 있다.

 

#7 요약

코루틴 빌더 launch, async, produce 각각의 반환형은 Unit, Deferred<T>, ReceiveChannel<E>다.