Coroutine Channels 3편 - Fan-out과 Fan-in : Channel이 얼마나 많은 출력, 입력을 만들 수 있는지 알아보기

Fan-out*1

복수의 Coroutine은 같은 채널로부터 수신하면서, 그들간에 작업을 분산할 수 있다. 1초에 10개의 숫자를 주기적으로 정수를 생성하는 생산자 Coroutine으로 시작하자 :

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1 // start from 1
    while (true) {
        send(x++) // produce next
        delay(100) // wait 0.1s
    }
}

그러면 우리는 몇개의 프로세서 Coroutine*2을 가질 수 있다. 이 예에서 프로세서 Coroutine은 그들의 id와 받은 숫자를 출력한다.

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
    for (msg in channel) {
        println("Processor #$id received $msg")
    }
}

이제 5개의 프로세서들을 실행하고 거의 1초간 동작하도록 하자. 어떤 일이 일어나는지 확인하자 :

val producer = produceNumbers()
repeat(5) { launchProcessor(it, producer) }
delay(950)
producer.cancel() // cancel producer coroutine and thus kill them all

📌 전체 코드는 이곳에서 확인할 수 있습니다.

특정 정수를 수신하는 프로세서의 id값이 다를 수 있지만, 출력은 다음과 비슷할 것이다.

Processor #2 received 1
Processor #4 received 2
Processor #0 received 3
Processor #1 received 4
Processor #3 received 5
Processor #2 received 6
Processor #4 received 7
Processor #0 received 8
Processor #1 received 9
Processor #3 received 10

생산자 Coroutine을 취소하는 것은 생산자 Coroutine의 채널을 닫는다. 그러므로 실제로 프로세서 Coroutine이 수행하는 채널의 반복을 종료한다.

또한, launchProcessor 코드에서 Fan-out을 수행하기 위해 명시적으로 for 루프를 사용해서 채널에 대해 반복을 수행한 방법에 주목하자. 이 for 루프 패턴은 consumeEach와 다르게 복수의 Coroutine에 사용하기 완전히 안전하다. 만약 하나의 프로세서 Coroutine이 실패하면 다른 Coroutine이 채널에 대한 처리를 할 것이다. 반면에, consumerEach를 사용해 작성된 프로세서는 정상적으로 혹은 비정상적으로 완료될 때 언제나 해당 채널을 소비(취소)한다.

📖 아래 내용은 독자의 이해를 위해 번역자가 추가한 글입니다.

*1. Fan-out이란 하나의 논리게이트의 출력이 얼마나 많은 논리게이트의 입력으로 사용되는지에 대한 것이다. 여기서는 Channel의 출력이 복수의 프로세서 Coroutine으로의 입력으로 사용되는 것을 뜻한다.

*2. 프로세서 Coroutine은 받은 값을 처리하는 처리기 Coroutine을 뜻한다.

이 글은 Coroutines 공식 문서를 번역한 글입니다.

원문 : Channels - Fan-out

원문 최종 수정 :  2022년 9월 28일

Fan-in*1

복수의 Coroutine이 같은 채널에 값을 보낼 수 있다. 예를 들어, 문자열을 다루는 채널이 있다고 해보고, 일정한 시간마다 채널로 특정한 문자열을 반복적으로 보내는 일시중단 함수가 있다고 해보자.

suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) {
    while (true) {
        delay(time)
        channel.send(s)
    }
}

이제, 문자열을 보내는 몇개의 Coroutine을 실행하면 어떤 일이 발생하는지 확인해보자(이 예에서는 메인 스레드의 Context에서 메인 Coroutine의 자식으로 Coroutine들을 실행한다) :

val channel = Channel<String>()
launch { sendString(channel, "foo", 200L) }
launch { sendString(channel, "BAR!", 500L) }
repeat(6) { // receive first six
    println(channel.receive())
}
coroutineContext.cancelChildren() // cancel all children to let main finish

📌 전체 코드는 이곳에서 확인할 수 있습니다.

출력은 다음과 같다 :

foo
foo
BAR!
foo
foo
BAR!

📖 아래 내용은 독자의 이해를 위해 번역자가 추가한 글입니다.

*1. Fan-in은 논리 게이트가 받을 수 있는 최대 입력 수이다. 이 글에서는 Channel에 입력하는 Coroutine을 여러 개 실행한다는 의미로 쓰인다.

이 글은 Coroutines 공식 문서를 번역한 글입니다.

원문 : Channels - Fan-in

원문 최종 수정 :  2022년 9월 28일

목차로 돌아가기