채널

채널 (Channels)

Deferred 값은 코루틴 사이에서 단일 값을 전달하는 편리한 방법을 제공해요. 채널(Channel)은 값의 스트림을 전달하는 방법을 제공합니다.

출처: Kotlin 공식 문서

본문

채널 기본

Channel은 개념적으로 BlockingQueue와 매우 비슷해요. 한 가지 핵심 차이는 블로킹 put 연산 대신 일시 중단(suspending)되는 send를 가지고, 블로킹 take 연산 대신 일시 중단되는 receive를 가진다는 점이에요.

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

fun main() = runBlocking {
//sampleStart
    val channel = Channel<Int>()
    launch {
        // this might be heavy CPU-consuming computation or async logic, 
        // we'll just send five squares
        for (x in 1..5) channel.send(x * x)
    }
    // here we print five received integers:
    repeat(5) { println(channel.receive()) }
    println("Done!")
//sampleEnd
}

전체 코드는 여기에서 가져올 수 있어요.

이 코드의 출력은 다음과 같습니다.

1
4
9
16
25
Done!

채널 닫기와 순회

큐와 달리 채널은 더 이상 요소가 오지 않음을 나타내기 위해 닫을 수 있어요. 수신 측에서는 일반 for 루프를 사용해 채널에서 요소를 받는 것이 편리합니다.

개념적으로 close는 특수한 close 토큰을 채널에 보내는 것과 같아요. 이 close 토큰을 받는 즉시 순회가 멈추므로, close 이전에 보낸 모든 요소는 반드시 수신된다는 보장이 있습니다.

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

fun main() = runBlocking {
//sampleStart
    val channel = Channel<Int>()
    launch {
        for (x in 1..5) channel.send(x * x)
        channel.close() // we're done sending
    }
    // here we print received values using `for` loop (until the channel is closed)
    for (y in channel) println(y)
    println("Done!")
//sampleEnd
}

전체 코드는 여기에서 가져올 수 있어요.

채널 프로듀서 만들기

코루틴이 요소의 시퀀스를 생산하는 패턴은 꽤 흔해요. 이는 동시성 코드에서 자주 볼 수 있는 producer-consumer 패턴의 일부입니다. 이런 프로듀서를 채널을 파라미터로 받는 함수로 추상화할 수도 있지만, 이는 결과가 함수에서 반환되어야 한다는 상식에 반하는 일이에요.

프로듀서 측에서 이를 올바르게 처리하기 쉽게 해 주는 produce라는 편리한 코루틴 빌더가 있고, 컨슈머 측에서 for 루프를 대체하는 consumeEach라는 확장 함수가 있어요.

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

//sampleStart
fun CoroutineScope.produceSquares(): ReceiveChannel<Int> = produce {
    for (x in 1..5) send(x * x)
}

fun main() = runBlocking {
    val squares = produceSquares()
    squares.consumeEach { println(it) }
    println("Done!")
//sampleEnd
}

전체 코드는 여기에서 가져올 수 있어요.

파이프라인 (Pipelines)

파이프라인은 한 코루틴이 (어쩌면 무한한) 값의 스트림을 생산하는 패턴이에요.

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++) // infinite stream of integers starting from 1
}

그리고 다른 코루틴(들)이 그 스트림을 소비해 일부 처리를 하고 다른 결과를 생산합니다. 아래 예시에서 숫자는 단순히 제곱됩니다.

fun CoroutineScope.square(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (x in numbers) send(x * x)
}

main 코드가 전체 파이프라인을 시작하고 연결해요.

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

fun main() = runBlocking {
//sampleStart
    val numbers = produceNumbers() // produces integers from 1 and on
    val squares = square(numbers) // squares integers
    repeat(5) {
        println(squares.receive()) // print first five
    }
    println("Done!") // we are done
    coroutineContext.cancelChildren() // cancel children coroutines
//sampleEnd
}

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++) // infinite stream of integers starting from 1
}

fun CoroutineScope.square(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (x in numbers) send(x * x)
}

전체 코드는 여기에서 가져올 수 있어요.

코루틴을 만드는 모든 함수는 CoroutineScope의 확장으로 정의되어 있어요. 그래서 구조적 동시성(structured concurrency)에 의존해 애플리케이션에 남아도는 전역 코루틴이 없도록 할 수 있습니다.

파이프라인으로 소수 구하기

코루틴 파이프라인으로 소수(prime number)를 생성하는 예시를 통해 파이프라인을 극한까지 사용해 볼게요. 무한 숫자 시퀀스에서 시작합니다.

fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
    var x = start
    while (true) send(x++) // infinite stream of integers from start
}

다음 파이프라인 스테이지는 들어오는 숫자 스트림을 필터링해 주어진 소수로 나누어떨어지는 모든 숫자를 제거해요.

fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
    for (x in numbers) if (x % prime != 0) send(x)
}

이제 2에서 시작하는 숫자 스트림을 시작하고, 현재 채널에서 소수를 가져오며, 찾은 각 소수마다 새 파이프라인 스테이지를 실행해 파이프라인을 구축합니다.

numbersFrom(2) -> filter(2) -> filter(3) -> filter(5) -> filter(7) ... 

다음 예시는 전체 파이프라인을 main 스레드의 컨텍스트에서 실행하며 처음 열 개의 소수를 출력해요. 모든 코루틴이 main runBlocking 코루틴의 스코프에서 실행되므로 시작한 모든 코루틴의 명시적 목록을 유지할 필요가 없습니다. 처음 열 개의 소수를 출력한 후 cancelChildren 확장 함수를 사용해 모든 자식 코루틴을 취소합니다.

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

fun main() = runBlocking {
//sampleStart
    var cur = numbersFrom(2)
    repeat(10) {
        val prime = cur.receive()
        println(prime)
        cur = filter(cur, prime)
    }
    coroutineContext.cancelChildren() // cancel all children to let main finish
//sampleEnd    
}

fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
    var x = start
    while (true) send(x++) // infinite stream of integers from start
}

fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
    for (x in numbers) if (x % prime != 0) send(x)
}

전체 코드는 여기에서 가져올 수 있어요.

이 코드의 출력은 다음과 같습니다.

2
3
5
7
11
13
17
19
23
29

표준 라이브러리의 iterator 코루틴 빌더로도 같은 파이프라인을 만들 수 있다는 점을 기억하세요. produce를 iterator로, send를 yield로, receive를 next로, ReceiveChannel을 Iterator로 바꾸고 코루틴 스코프를 제거하면 돼요. runBlocking도 필요하지 않습니다. 하지만 위에서 보여준 채널을 사용하는 파이프라인의 이점은 Dispatchers.Default 컨텍스트에서 실행하면 실제로 여러 CPU 코어를 사용할 수 있다는 점이에요.

어쨌든 이것은 소수를 찾는 매우 비현실적인 방법입니다. 실제로 파이프라인은 원격 서비스에 대한 비동기 호출 같은 다른 일시 중단 호출을 포함하기 마련인데, sequence/iterator로는 완전히 비동기인 produce와 달리 임의의 일시 중단을 허용하지 않기 때문에 이런 파이프라인을 만들 수 없어요.

Fan-out

여러 코루틴이 같은 채널에서 수신하면서 서로 작업을 분배할 수 있어요. 주기적으로 정수(초당 10개)를 생산하는 프로듀서 코루틴부터 시작해 볼게요.

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

그 다음 여러 프로세서 코루틴을 만들 수 있어요. 이 예시에서 그들은 자신의 id와 수신한 숫자만 출력합니다.

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

이제 5개의 프로세서를 실행하고 거의 1초 동안 작동시키고, 무슨 일이 일어나는지 봅시다.

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

fun main() = runBlocking<Unit> {
//sampleStart
    val producer = produceNumbers()
    repeat(5) { launchProcessor(it, producer) }
    delay(950)
    producer.cancel() // cancel producer coroutine and thus kill them all
//sampleEnd
}

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

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

전체 코드는 여기에서 가져올 수 있어요.

특정 정수를 받는 프로세서 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

프로듀서 코루틴을 취소하면 그 채널이 닫혀서, 프로세서 코루틴이 수행하던 채널 순회가 결국 종료된다는 점을 기억하세요.

또한 launchProcessor 코드에서 fan-out을 수행하기 위해 for 루프로 채널을 명시적으로 순회한다는 것에 주목하세요. consumeEach와 달리 이 for 루프 패턴은 여러 코루틴에서 사용해도 완벽히 안전합니다. 프로세서 코루틴 중 하나가 실패해도 나머지는 여전히 채널을 처리하지만, consumeEach로 작성된 프로세서는 정상 또는 비정상 완료 시 항상 기본 채널을 소비(취소)합니다.

Fan-in

여러 코루틴이 같은 채널로 보낼 수 있어요. 예를 들어 문자열 채널과, 지정된 지연으로 이 채널에 지정된 문자열을 반복해서 보내는 일시 중단 함수가 있다고 해 볼게요.

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

이제 문자열을 보내는 코루틴 몇 개를 실행하면(이 예시에서는 main 코루틴의 자식으로 main 스레드 컨텍스트에서 실행) 무슨 일이 일어나는지 봅시다.

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

fun main() = runBlocking {
//sampleStart
    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
//sampleEnd
}

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

전체 코드는 여기에서 가져올 수 있어요.

출력은 다음과 같습니다.

foo
foo
BAR!
foo
foo
BAR!

버퍼 채널 (Buffered channels)

지금까지 본 채널에는 버퍼가 없었어요. 버퍼가 없는 채널은 송신자와 수신자가 서로 만날 때(일명 rendezvous) 요소를 전송합니다. send가 먼저 호출되면 receive가 호출될 때까지 일시 중단되고, receive가 먼저 호출되면 send가 호출될 때까지 일시 중단돼요.

Channel() 팩토리 함수와 produce 빌더 모두 버퍼 크기를 지정하는 선택적 capacity 파라미터를 받아요. 버퍼는 송신자가 일시 중단되기 전에 여러 요소를 보낼 수 있게 해 주는데, 지정된 용량을 가진 BlockingQueue가 버퍼가 가득 차면 블로킹되는 것과 비슷합니다.

다음 코드의 동작을 살펴볼게요.

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

fun main() = runBlocking<Unit> {
//sampleStart
    val channel = Channel<Int>(4) // create buffered channel
    val sender = launch { // launch sender coroutine
        repeat(10) {
            println("Sending $it") // print before sending each element
            channel.send(it) // will suspend when buffer is full
        }
    }
    // don't receive anything... just wait....
    delay(1000)
    sender.cancel() // cancel sender coroutine
//sampleEnd    
}

전체 코드는 여기에서 가져올 수 있어요.

용량이 4인 버퍼 채널을 사용해 "sending"을 다섯 번 출력합니다.

Sending 0
Sending 1
Sending 2
Sending 3
Sending 4

첫 네 개의 요소는 버퍼에 추가되고, 송신자는 다섯 번째 요소를 보내려 할 때 일시 중단됩니다.

채널은 공정하다 (Channels are fair)

채널에 대한 send와 receive 연산은 여러 코루틴에서 호출되는 순서에 대해 공정합니다. 이들은 FIFO(선입선출) 순서로 처리되며, 예를 들어 첫 번째로 receive를 호출한 코루틴이 요소를 얻습니다. 다음 예시에서 "ping"과 "pong" 두 코루틴은 공유 "table" 채널에서 "ball" 객체를 수신하고 있어요.

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

//sampleStart
data class Ball(var hits: Int)

fun main() = runBlocking {
    val table = Channel<Ball>() // a shared table
    launch { player("ping", table) }
    launch { player("pong", table) }
    table.send(Ball(0)) // serve the ball
    delay(1000) // delay 1 second
    coroutineContext.cancelChildren() // game over, cancel them
}

suspend fun player(name: String, table: Channel<Ball>) {
    for (ball in table) { // receive the ball in a loop
        ball.hits++
        println("$name $ball")
        delay(300) // wait a bit
        table.send(ball) // send the ball back
    }
}
//sampleEnd

전체 코드는 여기에서 가져올 수 있어요.

"ping" 코루틴이 먼저 시작되므로 가장 먼저 ball을 받아요. "ping" 코루틴이 ball을 table로 다시 보낸 직후 곧바로 ball을 다시 받기 시작하지만, "pong" 코루틴이 이미 기다리고 있었기 때문에 ball은 "pong" 코루틴이 받게 됩니다.

ping Ball(hits=1)
pong Ball(hits=2)
ping Ball(hits=3)
pong Ball(hits=4)

사용 중인 실행자(executor)의 특성 때문에 채널이 때로 공정하지 않아 보이는 실행을 만들어 낼 수 있다는 점도 알아두세요. 자세한 내용은 이 이슈를 참고하세요.

티커 채널 (Ticker channels)

티커 채널은 이 채널에서 마지막으로 소비한 이후 주어진 지연이 지날 때마다 Unit을 생산하는 특수한 rendezvous 채널이에요. 단독으로는 쓸모없어 보일 수 있지만, 윈도우잉과 기타 시간 종속 처리를 수행하는 복잡한 시간 기반 produce 파이프라인과 연산자를 만드는 유용한 구성 요소입니다. 티커 채널은 select에서 "on tick" 동작을 수행하는 데 사용할 수 있어요.

이런 채널을 만들려면 팩토리 메서드 ticker를 사용해요. 더 이상 요소가 필요 없다는 것을 나타내려면 ReceiveChannel.cancel 메서드를 사용합니다.

이제 실제로 어떻게 동작하는지 볼게요.

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

//sampleStart
fun main() = runBlocking<Unit> {
    val tickerChannel = ticker(delayMillis = 200, initialDelayMillis = 0) // create a ticker channel
    var nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
    println("Initial element is available immediately: $nextElement") // no initial delay

    nextElement = withTimeoutOrNull(100) { tickerChannel.receive() } // all subsequent elements have 200ms delay
    println("Next element is not ready in 100 ms: $nextElement")

    nextElement = withTimeoutOrNull(120) { tickerChannel.receive() }
    println("Next element is ready in 200 ms: $nextElement")

    // Emulate large consumption delays
    println("Consumer pauses for 300ms")
    delay(300)
    // Next element is available immediately
    nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
    println("Next element is available immediately after large consumer delay: $nextElement")
    // Note that the pause between `receive` calls is taken into account and next element arrives faster
    nextElement = withTimeoutOrNull(120) { tickerChannel.receive() }
    println("Next element is ready in 100ms after consumer pause in 300ms: $nextElement")

    tickerChannel.cancel() // indicate that no more elements are needed
}
//sampleEnd

전체 코드는 여기에서 가져올 수 있어요.

다음 줄들을 출력합니다.

Initial element is available immediately: kotlin.Unit
Next element is not ready in 100 ms: null
Next element is ready in 200 ms: kotlin.Unit
Consumer pauses for 300ms
Next element is available immediately after large consumer delay: kotlin.Unit
Next element is ready in 100ms after consumer pause in 300ms: kotlin.Unit

ticker는 가능한 컨슈머 일시 중지를 인지하며, 기본적으로 일시 중지가 발생하면 다음 생산 요소의 지연을 조정해 생산 요소의 고정 비율을 유지하려 한다는 점을 기억하세요.

선택적으로 요소 사이의 고정 지연을 유지하려면 TickerMode.FIXED_DELAY와 같은 mode 파라미터를 지정할 수 있어요.

더 알아보기 (Learn more)

• Kotlin 공식 문서의 Channels 페이지
• Composing suspending functions · Coroutine context and dispatchers
• Coroutine exceptions handling