3.2 채널과 플로우 - 셀렉트

 

 

코루틴에는 select 함수가 있다. 이 select 함수는 가장 먼저 완료되는 코루틴의 결과를 기다리는 역할을 한다.

또한 여러 개의 채널 중 남은 공간이 있는 채널에 데이터를 보내거나, 여러 개의 채널 중 원소가 존재하는 채널로부터 원소를 받을 수도 있다.

코루틴 사이에 경합을 일으키거나, 여러 개의 데이터 소스로부터 나오는 결괏값을 합칠 수도 있다.

 

select 함수는 코틀린 코루틴이 출시된 이후부터 사용이 가능했지만, 아직 실험용이다. 즉, API 형태가 바뀔 가능성이 있다.

select를 실제로 사용하는 경우가 드물기 때문에, 안정화될 가능성도 적다.

 

 

지연되는 값 선택하기

여러 개의 소스에 데이터를 요청한 뒤, 가장 빠른 응답만 얻는 경우를 생각해 보자.

가장 쉬운 방법은 요청을 여러 개의 비동기 프로세스로 시작한 뒤, select 함수 내부에서 값을 기다리는 것이다.

select 내부에서는 람다 내 Deferred의 onAwait 함수를 호출한다. 또한 람다 내에서 값을 변환할 수도 있다.

 

suspend fun requestData1(): String {
    delay(100_000)
    return "Data1"

}

suspend fun requestData2(): String {
    delay(1000)
    return "Data2"
}

val scope = CoroutineScope(SupervisorJob())

suspend fun askMultipleForData(): String {
    val defData1 = scope.async { requestData1() }
    val defData2 = scope.async { requestData2() }

    return select {
        defData1.onAwait { it }
        defData2.onAwait { it }
    }
}

suspend fun main(): Unit = coroutineScope {
    println(askMultipleForData())
}

(1초 후)
Data2

하나의 비동기 작업이 먼저 완료되면 select 함수도 바로 끝나기 때문에, 하나의 결괏값만 반환된다.

 

위 예제에서는 외부의 스코프에서 async 코루틴이 시작된다.

즉, askMultipleForData를 호출한 코루틴을 취소해도, 외부 스코프에서 시작된 async 코루틴은 취소되지 않는다.

 

suspend fun askMultipleForData(): String = coroutineScope {
    select<String> {
        async { requestData1() }.onAwait { it }
        async { requestData2() }.onAwait { it }
    }
}

suspend fun main(): Unit = coroutineScope {
    println(askMultipleForData())
}

(100초 후)
Data2

coroutineScope를 사용해서 비동기 작업과 select를 하나의 스코프에서 시작시키는 예제를 보자.

스코프는 모든 자식 코루틴이 끝날 때까지 기다리기 때문에, 1초가 아닌 100초 후에서야 Data2를 결과로 받는다.

 

suspend fun askMultipleForData(): String = coroutineScope {
    select<String> {
        async { requestData1() }.onAwait { it }
        async { requestData2() }.onAwait { it }
    }.also { coroutineContext.cancelChildren() }
}

suspend fun main(): Unit = coroutineScope {
    println(askMultipleForData())
}

(1초 후)
Data2

100초 후 Data2를 결과로 받는 문제를 해결하기 위해, 위처럼 코드를 수정했다.

select가 값을 생성하고 난 후, also 블럭 내부에서 자식 코루틴을 취소시킨다.

이처럼 async와 select를 사용하면 코루틴끼리 경합하는 상황을 쉽게 구현할 수 있지만, 스코프를 명시적으로 취소해야 한다.

 

suspend fun askMultipleForData(): String = raceOf(
    { requestData1() },
    { requestData2() },
)

suspend fun main(): Unit = coroutineScope {
    println(askMultipleForData())
}

(1초 후)
Data2

스코프를 명시적으로 취소하는 방식은 약간 복잡하기 때문에, 헬퍼 함수를 정의하거나 raceOf 함수를 지원하는 외부 라이브러리를 사용할 수 있다.

추후 raceOf 함수를 직접 구현해 볼 것이다.

위 예제에서는 Splitties 라이브러리의 raceOf를 사용했다.

 

 

채널에서 값 선택하기

select 함수는 채널에서도 사용할 수 있다. select 람다 내에서 사용하는 주요 함수는 다음과 같다.

• onReceive
  • 채널이 값을 가지고 있을 때 선택된다.
  • (receive 메서드처럼) 값을 받은 뒤 람다식의 인자로 사용한다.
  • select는 람다식의 결괏값을 반환한다.
• onReceiveCatching
  • 채널이 값을 가지고 있거나 닫혔을 때 선택된다.
  • 값을 나타내거나 채널이 닫혔다는 걸 알려주는 ChannelResult를 받아 람다식의 인자로 사용한다.
  • select는 람다식의 결괏값을 반환한다.
• onSend
  • 채널의 버퍼에 공간이 있을 때 선택된다.
  • (send 메서드처럼) 채널에 값을 보낸 뒤, 채널의 참조값으로 람다식을 수행한다.
  • select는 Unit을 반환한다.

 

suspend fun CoroutineScope.produceString(
    s: String,
    time: Long
) = produce {
    while (true) {
        delay(time)
        send(s)
    }
}

fun main() = runBlocking {
    val fooChannel = produceString("foo", 210L)
    val barChannel = produceString("BAR", 500L)

    repeat(7) {
        select {
            fooChannel.onReceive {
                println("From fooChannel: $it")
            }
            barChannel.onReceive {
                println("From barChannel: $it")
            }
        }
    }

    coroutineContext.cancelChildren()
}

From fooChannel: foo
From fooChannel: foo
From barChannel: BAR
From fooChannel: foo
From fooChannel: foo
From barChannel: BAR
From fooChannel: foo

select 람다 내부에서는 여러 개의 채널로부터 결괏값을 얻기 위해 onReceive나 onReceiveCatching을 함께 사용한다.

 

fun main(): Unit = runBlocking {
    val c1 = Channel<Char>(capacity = 2)
    val c2 = Channel<Char>(capacity = 2)

    // 값을 보냄
    launch {
        for (c in 'A'..'H') {
            delay(400)
            select<Unit> {
                c1.onSend(c) { println("Sent $c to 1") }
                c2.onSend(c) { println("Sent $c to 2") }
            }
        }
    }

    // 값을 받음
    launch {
        while (true) {
            delay(1000)
            val c = select<String> {
                c1.onReceive { "$it from 1" }
                c2.onReceive { "$it from 2" }
            }
            println("Received $c")
        }
    }
}

Sent A to 1
Sent B to 1
Received A from 1
Sent C to 1
Sent D to 2
Received B from 1
Sent E to 1
Sent F to 2
Received C from 1
Sent G to 1
Received E from 1
Sent H to 1
Received G from 1
Received H from 1
Received D from 2
Received F from 2

select 함수에서 onSend를 호출하면 버퍼에 공간이 있는 채널에 원소를 보내는 용도로 사용할 수 있다.

 

 

요약

select는 가장 먼저 완료되는 코루틴의 결괏값을 기다릴 때나, 여러 개의 채널 중 전송 또는 수신 가능한 채널을 선택할 때 유용하다.

주로 채널과 함께 사용하지만, async 코루틴의 경합을 구현하기 위해 사용할 수도 있다.

 

 

 

출처

https://www.yes24.com/product/goods/123034354