코루틴 Select 표현식 (Coroutines Select expression)
Select 표현식은 여러 중단 함수를 동시에 기다릴 수 있게 하고 첫 번째 사용 가능한 것을 선택할 수 있게 합니다.
Select 표현식은 kotlinx.coroutines의 실험적인 기능입니다. 해당 API는 kotlinx.coroutines 라이브러리의 업데이트 중에 변경될 수 있는 손상을 일으킬 수 있는 변경 사항이 예상됩니다.
채널에서 선택하기
우리는 두 개의 문자열 프로듀서, "Fizz"와 "Buzz!"를 가지고 있습니다. Fizz는 500ms마다 "Fizz" 문자열을 생성합니다:
fun CoroutineScope.fizz() = produce<String> {
while (true) { // 500ms마다 "Fizz"를 전송
delay(500)
send("Fizz")
}
}
Buzz는 1000ms마다 "Buzz!" 문자열을 생성합니다:
fun CoroutineScope.buzz() = produce<String> {
while (true) { // 1000ms마다 "Buzz!"를 전송
delay(1000)
send("Buzz!")
}
}
receive 중단 함수를 사용하여 하나의 채널 또는 다른 채널에서 수신할 수 있습니다. 그러나 select 표현식은 onReceive 절을 사용하여 onReceive 절을 사용하여 동시에 두 개의 채널에서 수신할 수 있게 합니다:
suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
select<Unit> { // <Unit>은 이 select 표현식이 결과를 생성하지 않는다는 의미입니다.
fizz.onReceive { value -> // 이것은 첫 번째 select 절입니다.
println("fizz -> '$value'")
}
buzz.onReceive { value -> // 이것은 두 번째 select 절입니다.
println("buzz -> '$value'")
}
}
}
이를 7번 실행해 봅시다:
val fizz = fizz()
val buzz = buzz()
repeat(7) {
selectFizzBuzz(fizz, buzz)
}
coroutineContext.cancelChildren() // fizz 및 buzz 코루틴 취소
이 코드의 결과는 다음과 같습니다:
fizz -> 'Fizz'
buzz -> 'Buzz!'
fizz -> 'Fizz'
fizz -> 'Fizz'
buzz -> 'Buzz!'
fizz -> 'Fizz'
fizz -> 'Fizz'
닫힌 상태에서 선택하기
선택된 채널에서 onReceive 절은 채널이 닫힐 때 실패하고 해당 select에서 예외를 throw합니다. 채널이 닫힐 때 특정 작업을 수행하려면 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 {
"채널 'a'가 닫혔습니다."
}
}
b.onReceiveCatching { it ->
val value = it.getOrNull()
if (value != null) {
"b -> '$value'"
} else {
"채널 'b'가 닫혔습니다."
}
}
}
채널 a는 "Hello" 문자열을 네 번 생성하고 채널 b는 "World" 문자열을 네 번 생성합니다:
val a = produce<String> {
repeat(4) { send("Hello $it") }
}
val b = produce<String> {
repeat(4) { send("World $it") }
}
repeat(8) { // 처음 여덟 결과를 출력
println(selectAorB(a, b))
}
coroutineContext.cancelChildren()
이 코드의 결과는 다소 흥미로운데, 자세히 분석하겠습니다:
a -> 'Hello 0'
a -> 'Hello 1'
b -> 'World 0'
a -> 'Hello 2'
a -> 'Hello 3'
b -> 'World 1'
채널 'a'가 닫혔습니다.
채널 'a'가 닫혔습니다.
두 가지 관찰 사항을 얻을 수 있습니다.
먼저, select는 첫 번째 절에 편향됩니다. 동시에 여러 절이 선택 가능할 때, 그 중 첫 번째 절이 선택됩니다. 여기에서 두 채널 모두 계속 문자열을 생성하므로 첫 번째 절인 a 채널이 선택됩니다. 그러나 버퍼가 없는 채널을 사용하므로 a는 때때로 send 호출에서 중단되어 b도 전송할 기회를 얻습니다.
두 번째 관찰은 채널이 이미 닫힌 경우에 onReceiveCatching이 즉시 선택된다는 것입니다.
전송 선택
Select 표현식은 onSend 절을 가지고 있으며 이것은 선택의 편향된 성격과 함께 사용할 때 큰 도움이 될 수 있습니다.
우선 소비자가 주 채널에서 처리하지 못할 때 정수 값을 사이드 채널로 보내는 정수 프로듀서의 예를 작성해 보겠습니다:
fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
for (num in 1..10) { // 1에서 10까지 10개의 숫자를 생성
delay(100) // 100ms마다
select<Unit> {
onSend(num) {} // 주 채널로 보내기
side.onSend(num) {} // 또는 사이드 채널로 보내기
}
}
}
소비자는 매우 느린데, 각 숫자를 처리하는 데 250ms가 걸립니다:
val side = Channel<Int>() // 사이드 채널 할당
launch { // 이것은 사이드 채널에 대한 매우 빠른 소비자입니다.
side.consumeEach { println("사이드 채널에 $it이 있습니다.") }
}
produceNumbers(side).consumeEach {
println("소비 중: $it")
delay(250) // 소비된 숫자를 제대로 소화하기 위해 250ms 지연
}
println("소비 완료")
coroutineContext.cancelChildren()
그러면 무슨 일이 벌어지는지 살펴보겠습니다:
소비 중: 1
사이드 채널에 2이 있습니다.
사이드 채널에 3이 있습니다.
소비 중: 4
사이드 채널에 5이 있습니다.
사이드 채널에 6이 있습니다.
소비 중: 7
사이드 채널에 8이 있습니다.
사이드 채널에 9이 있습니다.
소비 중: 10
소비 완료
채널에서 선택하기
Deferred 값은 onAwait 절을 사용하여 선택할 수 있습니다. 임의의 지연 값을 가진 문자열 값을 반환하는 async 함수부터 시작해 보겠습니다:
fun CoroutineScope.asyncString(time: Int) = async {
delay(time.toLong())
"$time ms를 기다렸습니다"
}
랜덤한 지연 시간으로 12개의 값을 시작해 보겠습니다.
fun CoroutineScope.asyncStringsList(): List<Deferred<String>> {
val random = Random(3)
return List(12) { asyncString(random.nextInt(1000)) }
}
이제 주 함수는 첫 번째로 완료된 것을 기다리고 아직 활성 상태인 지연된 값의 수를 계산합니다. 여기서는 select 표현식이 Kotlin DSL임을 활용하여 임의의 코드를 사용하여 이를 제공할 수 있음을 주목하세요. 이 경우 지연된 값 목록을 반복하여 각 지연 값에 대한 onAwait 절을 제공합니다.
val list = asyncStringsList()
val result = select<String> {
list.withIndex().forEach { (index, deferred) ->
deferred.onAwait { answer ->
"Deferred $index이(가) '$answer'을(를) 생성했습니다"
}
}
}
println(result)
val countActive = list.count { it.isActive }
println("$countActive 개의 코루틴이 아직 활성 상태입니다")
출력은 다음과 같습니다:
Deferred 4이(가) 'Waited for 128 ms'를(을) 생성했습니다
11 개의 코루틴이 아직 활성 상태입니다
Deferred 값의 채널을 사용해 전환하기
Deferred 문자열 값의 채널 프로듀서 함수를 작성하여 채널에서 지연된 문자열 값을 받고 다음 지연된 값이 올 때까지 또는 채널이 닫힐 때까지 기다리는 것을 구현할 수 있습니다. 이 예제에서는 동일한 select에서 onReceiveCatching 및 onAwait 절을 결합합니다:
fun CoroutineScope.switchMapDeferreds(input: ReceiveChannel<Deferred<String>>) = produce<String> {
var current = input.receive() // 첫 번째로 받은 지연된 값으로 시작
while (isActive) { // 취소/닫히지 않은 동안 루프 실행
val next = select<Deferred<String>?> { // 이 select에서 다음 지연된 값을 반환하거나 null을 반환
input.onReceiveCatching { update ->
update.getOrNull()
}
current.onAwait { value ->
send(value) // 현재 지연된 값이 생성한 값을 보냅니다.
input.receiveCatching().getOrNull() // 입력 채널에서 다음 지연된 값을 사용합니다.
}
}
if (next == null) {
println("채널이 닫혔습니다.")
break // 루프를 빠져나갑니다.
} else {
current = next
}
}
}
이를 테스트하기 위해 지정된 시간 후에 특정 문자열로 해결되는 간단한 async 함수를 사용합니다:
fun CoroutineScope.asyncString(str: String, time: Long) = async {
delay(time)
str
}
주 함수는 switchMapDeferreds의 결과를 인쇄하는 코루틴을 시작하고 테스트 데이터를 보내기 위해 몇 가지 테스트 데이터를 보냅니다:
val chan = Channel<Deferred<String>>() // 테스트용 채널
launch { // 인쇄 코루틴 시작
for (s in switchMapDeferreds(chan))
println(s) // 각 받은 문자열을 인쇄
}
chan.send(asyncString("시작", 100))
delay(200) // "시작"을 생성하는 데 충분한 시간
chan.send(asyncString("느림", 500))
delay(100) // 느림 생성에 충분한 시간이 아님
chan.send(asyncString("교체", 100))
delay(500) // 마지막 하나 전에 충분한 시간을 줌
chan.send(asyncString("끝", 500))
delay(1000) // 처리하는 데 충분한 시간을 줌
chan.close() // 채널 닫기...
delay(500) // 완료되도록 시간을 줌
이 코드의 결과는 다음과 같습니다:
시작
교체
끝
채널이 닫혔습니다.
이렇게 하여 select 표현식을 사용하여 코루틴 간에 통신하고 조절할 수 있는 다양한 상황을 다룰 수 있습니다.
원문
댓글