Select 표현식 (실험적

[//]: # (title: Select 표현식 (실험적))

Select 표현식을 사용하면 여러 개의 suspend 함수를 동시에 기다리면서 사용 가능한 첫 번째 함수를 _선택_할 수 있습니다.

Select 표현식은 kotlinx.coroutines의 실험적 기능입니다. 해당 API는 향후 kotlinx.coroutines 라이브러리 업데이트에서 잠재적으로 호환성을 깨는 변경과 함께 발전할 것으로 예상됩니다.

채널에서 선택하기

문자열을 생성하는 두 개의 프로듀서인 fizz와 buzz가 있다고 가정해 봅시다. fizz는 500ms마다 "Fizz" 문자열을 생성합니다:

fun CoroutineScope.fizz() = produce<String> {
    while (true) { // sends "Fizz" every 500 ms
        delay(500)
        send("Fizz")
    }
}

그리고 buzz는 1000ms마다 "Buzz!" 문자열을 생성합니다:

fun CoroutineScope.buzz() = produce<String> {
    while (true) { // sends "Buzz!" every 1000 ms
        delay(1000)
        send("Buzz!")
    }
}

[receive][ReceiveChannel.receive] suspend 함수를 사용하면 한 채널 또는 다른 채널 중 _하나_로부터 수신할 수 있습니다. 하지만 [select] 표현식은 [onReceive][ReceiveChannel.onReceive] 절을 사용하여 둘 다 동시에 수신할 수 있도록 해줍니다:

suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
    select<Unit> { // <Unit> means that this select expression does not produce any result 
        fizz.onReceive { value ->  // this is the first select clause
            println("fizz -> '$value'")
        }
        buzz.onReceive { value ->  // this is the second select clause
            println("buzz -> '$value'")
        }
    }
}

이를 총 7번 실행해 봅시다:

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

fun CoroutineScope.fizz() = produce<String> {
    while (true) { // sends "Fizz" every 500 ms
        delay(500)
        send("Fizz")
    }
}

fun CoroutineScope.buzz() = produce<String> {
    while (true) { // sends "Buzz!" every 1000 ms
        delay(1000)
        send("Buzz!")
    }
}

suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
    select<Unit> { // <Unit> means that this select expression does not produce any result 
        fizz.onReceive { value ->  // this is the first select clause
            println("fizz -> '$value'")
        }
        buzz.onReceive { value ->  // this is the second select clause
            println("buzz -> '$value'")
        }
    }
}

fun main() = runBlocking<Unit> {
    val fizz = fizz()
    val buzz = buzz()
    repeat(7) {
        selectFizzBuzz(fizz, buzz)
    }
    coroutineContext.cancelChildren() // cancel fizz & buzz coroutines
}

전체 코드는 여기에서 확인할 수 있습니다.

이 코드의 결과는 다음과 같습니다:

fizz -> 'Fizz'
buzz -> 'Buzz!'
fizz -> 'Fizz'
fizz -> 'Fizz'
buzz -> 'Buzz!'
fizz -> 'Fizz'
fizz -> 'Fizz'

채널 닫힘 시 선택하기

select의 [onReceive][ReceiveChannel.onReceive] 절은 채널이 닫히면 실패하여 해당 select가 예외를 발생시킵니다. 채널이 닫혔을 때 특정 작업을 수행하기 위해 [onReceiveCatching][ReceiveChannel.onReceiveCatching] 절을 사용할 수 있습니다. 다음 예제는 또한 select가 선택된 절의 결과를 반환하는 표현식임을 보여줍니다:

suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
    select<String> {
        a.onReceiveCatching { it ->
            val value = it.getOrNull()
            if (value != null) {
                "a -> '$value'"
            } else {
                "Channel 'a' is closed"
            }
        }
        b.onReceiveCatching { it ->
            val value = it.getOrNull()
            if (value != null) {
                "b -> '$value'"
            } else {
                "Channel 'b' is closed"
            }
        }
    }

"Hello" 문자열을 네 번 생성하는 채널 a와 "World"를 네 번 생성하는 채널 b와 함께 사용해 봅시다:

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

suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
    select<String> {
        a.onReceiveCatching { it ->
            val value = it.getOrNull()
            if (value != null) {
                "a -> '$value'"
            } else {
                "Channel 'a' is closed"
            }
        }
        b.onReceiveCatching { it ->
            val value = it.getOrNull()
            if (value != null) {
                "b -> '$value'"
            } else {
                "Channel 'b' is closed"
            }
        }
    }
    
fun main() = runBlocking<Unit> {
    val a = produce<String> {
        repeat(4) { send("Hello $it") }
    }
    val b = produce<String> {
        repeat(4) { send("World $it") }
    }
    repeat(8) { // print first eight results
        println(selectAorB(a, b))
    }
    coroutineContext.cancelChildren()  
}

전체 코드는 여기에서 확인할 수 있습니다.

이 코드의 결과는 매우 흥미로우므로, 더 자세히 분석해 보겠습니다:

a -> 'Hello 0'
a -> 'Hello 1'
b -> 'World 0'
a -> 'Hello 2'
a -> 'Hello 3'
b -> 'World 1'
Channel 'a' is closed
Channel 'a' is closed

여기에서 몇 가지를 관찰할 수 있습니다.

먼저, select는 첫 번째 절에 편향되어 있습니다. 여러 절이 동시에 선택 가능한 경우, 그중 첫 번째 절이 선택됩니다. 여기서는 두 채널 모두 지속적으로 문자열을 생성하므로, select의 첫 번째 절인 a 채널이 우선합니다. 하지만 버퍼링되지 않은 채널을 사용하기 때문에 a는 [send][SendChannel.send] 호출에서 때때로 suspend되고, b도 전송할 기회를 얻습니다.

두 번째 관찰은 채널이 이미 닫혔을 때 [onReceiveCatching][ReceiveChannel.onReceiveCatching]이 즉시 선택된다는 것입니다.

전송을 위한 선택

Select 표현식에는 선택의 편향된 특성과 조합하여 매우 유용하게 사용될 수 있는 [onSend][SendChannel.onSend] 절이 있습니다.

기본 채널의 소비자가 속도를 따라가지 못할 때 자신의 값을 side 채널로 전송하는 정수 프로듀서의 예제를 작성해 봅시다:

fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
    for (num in 1..10) { // produce 10 numbers from 1 to 10
        delay(100) // every 100 ms
        select<Unit> {
            onSend(num) {} // Send to the primary channel
            side.onSend(num) {} // or to the side channel     
        }
    }
}

소비자는 각 숫자를 처리하는 데 250ms가 걸리므로 꽤 느릴 것입니다:

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

fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
    for (num in 1..10) { // produce 10 numbers from 1 to 10
        delay(100) // every 100 ms
        select<Unit> {
            onSend(num) {} // Send to the primary channel
            side.onSend(num) {} // or to the side channel     
        }
    }
}

fun main() = runBlocking<Unit> {
    val side = Channel<Int>() // allocate side channel
    launch { // this is a very fast consumer for the side channel
        side.consumeEach { println("Side channel has $it") }
    }
    produceNumbers(side).consumeEach { 
        println("Consuming $it")
        delay(250) // let us digest the consumed number properly, do not hurry
    }
    println("Done consuming")
    coroutineContext.cancelChildren()  
}

전체 코드는 여기에서 확인할 수 있습니다.

그럼 어떤 일이 일어나는지 봅시다:

Consuming 1
Side channel has 2
Side channel has 3
Consuming 4
Side channel has 5
Side channel has 6
Consuming 7
Side channel has 8
Side channel has 9
Consuming 10
Done consuming

지연된 값 선택하기

Deferred 값은 [onAwait][Deferred.onAwait] 절을 사용하여 선택될 수 있습니다. 임의의 지연 후에 deferred 문자열 값을 반환하는 async 함수부터 시작해 봅시다:

fun CoroutineScope.asyncString(time: Int) = async {
    delay(time.toLong())
    "Waited for $time ms"
}

임의의 지연으로 12개의 async 함수를 시작해 봅시다.

fun CoroutineScope.asyncStringsList(): List<Deferred<String>> {
    val random = Random(3)
    return List(12) { asyncString(random.nextInt(1000)) }
}

이제 main 함수는 그중 첫 번째가 완료될 때까지 기다리고, 여전히 활성 상태인 deferred 값의 수를 계산합니다. select 표현식이 Kotlin DSL이라는 점을 여기서 활용했으므로, 임의의 코드를 사용하여 select 절을 제공할 수 있다는 점을 유의하세요. 이 경우, 각 deferred 값을 위한 onAwait 절을 제공하기 위해 deferred 값 리스트를 순회합니다.

import kotlinx.coroutines.*
import kotlinx.coroutines.selects.*
import java.util.*
    
fun CoroutineScope.asyncString(time: Int) = async {
    delay(time.toLong())
    "Waited for $time ms"
}

fun CoroutineScope.asyncStringsList(): List<Deferred<String>> {
    val random = Random(3)
    return List(12) { asyncString(random.nextInt(1000)) }
}

fun main() = runBlocking<Unit> {
    val list = asyncStringsList()
    val result = select<String> {
        list.withIndex().forEach { (index, deferred) ->
            deferred.onAwait { answer ->
                "Deferred $index produced answer '$answer'"
            }
        }
    }
    println(result)
    val countActive = list.count { it.isActive }
    println("$countActive coroutines are still active")
}

전체 코드는 여기에서 확인할 수 있습니다.

출력은 다음과 같습니다:

Deferred 4 produced answer 'Waited for 128 ms'
11 coroutines are still active

지연된 값 채널 전환하기

deferred 문자열 값 채널을 소비하고, 수신된 각 deferred 값을 기다리되, 다음 deferred 값이 도착하거나 채널이 닫힐 때까지만 기다리는 채널 프로듀서 함수를 작성해 봅시다. 이 예제는 동일한 select 내에서 [onReceiveCatching][ReceiveChannel.onReceiveCatching]와 [onAwait][Deferred.onAwait] 절을 함께 사용합니다:

fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> {
    var current = input.receive() // start with first received deferred value
    while (isActive) { // loop while not cancelled/closed
        val next = select<Deferred<String>?> { // return next deferred value from this select or null
            input.onReceiveCatching { update ->
                update.getOrNull()
            }
            current.onAwait { value ->
                send(value) // send value that current deferred has produced
                input.receiveCatching().getOrNull() // and use the next deferred from the input channel
            }
        }
        if (next == null) {
            println("Channel was closed")
            break // out of loop
        } else {
            current = next
        }
    }
}

이를 테스트하기 위해, 지정된 시간 후에 지정된 문자열로 해석되는 간단한 async 함수를 사용할 것입니다:

fun CoroutineScope.asyncString(str: String, time: Long) = async {
    delay(time)
    str
}

main 함수는 switchMapDeferreds의 결과를 출력하는 코루틴을 시작하고, 일부 테스트 데이터를 보냅니다:

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.selects.*
    
fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> {
    var current = input.receive() // start with first received deferred value
    while (isActive) { // loop while not cancelled/closed
        val next = select<Deferred<String>?> { // return next deferred value from this select or null
            input.onReceiveCatching { update ->
                update.getOrNull()
            }
            current.onAwait { value ->
                send(value) // send value that current deferred has produced
                input.receiveCatching().getOrNull() // and use the next deferred from the input channel
            }
        }
        if (next == null) {
            println("Channel was closed")
            break // out of loop
        } else {
            current = next
        }
    }
}

fun CoroutineScope.asyncString(str: String, time: Long) = async {
    delay(time)
    str
}

fun main() = runBlocking<Unit> {
    val chan = Channel<Deferred<String>>() // the channel for test
    launch { // launch printing coroutine
        for (s in switchMapDeferreds(chan)) 
            println(s) // print each received string
    }
    chan.send(asyncString("BEGIN", 100))
    delay(200) // enough time for "BEGIN" to be produced
    chan.send(asyncString("Slow", 500))
    delay(100) // not enough time to produce slow
    chan.send(asyncString("Replace", 100))
    delay(500) // give it time before the last one
    chan.send(asyncString("END", 500))
    delay(1000) // give it time to process
    chan.close() // close the channel ... 
    delay(500) // and wait some time to let it finish
}

전체 코드는 여기에서 확인할 수 있습니다.

이 코드의 결과:

BEGIN
Replace
END
Channel was closed