16장 채널 | Notion

코루틴끼리의 통신을 위한 기본적인 방법으로 채널 API가 추가되었다. 채널은 공공 책장에 비유할 수 있다. 다른 사람이 찾는 책을 한 사람이 먼저 가지고 와야한다. kotlinx.corutines의 Channel이 작동하는 방식과 비슷하다.

채널은 송신자와 수신자의 수에 제한이 없으며, 채널을 통해 전송된 모든 값은 단 한번만 받을 수 있다.

제목 없음.png

Channel은 두 개의 서로 다른 인터페이스를 구현한 하나의 인터페이스다.

SendChannel은 원소를 보내거나(또는 더하거나) 채널을 닫는 용도로 사용된다.
ReceiveChannel은 원소를 받을 때(또는 꺼낼 때) 사용된다.

import kotlinx.coroutines.CancellationException

interface SendChannel<in E> {
    suspend fun send(element: E)
    fun close(): Boolean
    // ..
}

interface ReceiveChannel<out E> {
    suspend fun receive(): E
    fun cancel(cause: CancellationException? = null)
    // ..
}

두 인터페이스는 구분되어 있으며, 채널의 진입점을 제한하기 위해 Receive Channel이나 SendChannel 중 하나만 노출시키는 것도 가능하다.

send와 receive 모두 중단 함수라는 것을 확인할 수 있다. 원소를 보내고 받는 함수가 중단 함수인 것은 필수적인 특징이다.

receive를 호출했는데 채널에 원소가 없다면 코루틴은 원소가 들어올 때까지 중단된다. 공유 책장으로 비유하면, 누군가 책을 찾으러 갔는데 책장이 비어 있는 경우, 다른 사람이 책을 넣을 때까지 기다려야 하는 상황이다.
반면 send는 채널의 용량이 다 찼을 때 중단된다. 대부분의 채널은 용량이 제한되어 있다는 걸 나중에 확인할 수 있다. 공유 책장으로 비유하면, 누군가 책을 넣으러 갔는데 책장이 가득 찬 경우, 다른 사람이 책을 가져가 공간이 생길 때까지 기다려야 하는 상황이다.

중단 함수가 아닌 함수로 보내거나 받아야 한다면 trySend와 tryReceive를 사용할 수 있다. 두 연산 모두 연산이 성공했는지 실패했는지에 대한 정보를 담고 있는 ChannelResult를 즉시 반환한다. 용량이 제한적인 채널에서만 trySend와 tryReceive를 사용해야 하는데, (버퍼가 없는) 랑데뷰 채널(Channel.RENDEZVOUS)에서는 작동하지 않기 때문이다.

채널은 송신자와 수신자의 수에 제한이 없다. 하지만 채널의 양쪽 끝에 각각 하나의 코루틴만 있는 경우가 일반적이다.

채널의 가장 간단한 예를 보려면 각기 다른 코루틴에 생성자(송신자)와 소비자(수신자)가 있어야 한다. 생성자는 원소를 보내고 소비자는 원소를 받는다. 다음은 이러한 상황을 구현한 예이다.

suspend fun main(): Unit = coroutineScope {
    val channel = Channel<Int>()

    launch {
        repeat(5) { index ->
            delay(1000)
            println("Producing next one")
            channel.send(index * 2)
        }
    }

    launch {
        repeat(5) {
            val received = channel.receive()
            println(received)
        }
    }
}

Producing next one
0
Producing next one
2
Producing next one
4
Producing next one
6
Producing next one
8

위와 같은 구현 방식은 불완전하다. 우선 수신자는 얼마나 많은 원소를 보내는지 알아야 한다(여기서는 5라고 명시함). 수신자가 이런 정보를 아는 경우는 별로 없기 때문에 송신자가 보내는 만큼 수신자가 기다리는 방식을 선호한다. 채널이 닫힐 때까지 원소를 받기 위해 for 루프나 consumeEach 함수를 사용할 수 있다.( consumeEach 함수 또한 for 루프를 사용하지만 모든 원소를 가지고 온 다음에 (채널이 닫힌 뒤) 채널을 취소한다는 차이가 있다. )

suspend fun main(): Unit = coroutineScope {
    val channel = Channel<Int>()

    launch {
        repeat(5) { index ->
            println("Producing next one")
            delay(1000)
            channel.send(index * 2)
        }
        channel.close()
    }

    launch {
        for (element in channel) {
            println(element)
        }
//        channel.consumeEach { element ->
//            println(element)
//        }
    }
}

Producing next one
Producing next one
0
Producing next one
2
Producing next one
4
Producing next one
6
8