四種不同的 Channel

Channel Builder 的參數#

上一篇我們直接用 Channel<E>() 建立通道，而這個函式其實有三個參數：

public fun <E> Channel(
    capacity: Int = RENDEZVOUS,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    onUndeliveredElement: ((E) -> Unit)? = null
): Channel<E>

預設情況下：

capacity = RENDEZVOUS
onBufferOverflow = BufferOverflow.SUSPEND
onUndeliveredElement = null

不同的 capacity 會帶來不同的行為，這也是 Channel 主要的分類依據。

Capacity 的五種設定#

Channel.Capacity 大致可以分成五類：

RENDEZVOUS
CONFLATED
UNLIMITED
BUFFERED
其他（自訂正整數）

Rendezvous Channel#

預設值，沒有 Buffer。send 與 receive 必須同時出現，元素才會交換：

只有 send 被呼叫、沒有 receive：send 會暫停。
只有 receive 被呼叫、沒有 send：receive 會暫停。

換句話說，兩者是「成雙成對」的關係。從原始碼可以看到 RENDEZVOUS 的特殊處理：

RENDEZVOUS -> {
    if (onBufferOverflow == BufferOverflow.SUSPEND)
        RendezvousChannel(onUndeliveredElement)
    else
        ArrayChannel(1, onBufferOverflow, onUndeliveredElement)
}

只有在 onBufferOverflow 為 BufferOverflow.SUSPEND（也就是預設值）時，才會建立真正的 RendezvousChannel；否則會退化成 capacity 為 1 的 ArrayChannel。

Conflated Channel#

Conflated Channel（合併的通道）並不是把元素累加合併起來，而是只保留「最新一個」。它的 Buffer 大小為 1，但下一個 send 進來時，舊值會被覆蓋。

class Day16 {
    val scope = CoroutineScope(Job())

    suspend fun conflatedBroadcastChannel(): Unit {
        val channel = Channel<Int>(capacity = Channel.CONFLATED)
        scope.launch {
            for (i in 0..10) {
                channel.send(i)
            }

            delay(1000)
            channel.send(9)
        }

        for (i in 0..100) {
            delay(100)
            println(channel.receive())
        }

        println("done")
    }
}

範例輸出：

10
9

說明：

第一段 send 連續送出 0~10，但 receive 比較慢（每 100 毫秒才取一次），所以前面的值都被新值覆蓋，最後只剩 10。
等待 1 秒後又呼叫 send(9)，這個 9 落在 receive 的 for-loop 範圍內，所以也會被印出。
後面 receive 沒有新資料可取，就會持續暫停。

從 Channel builder 的內部實作來看，CONFLATED 等同於設定 onBufferOverflow = DROP_OLDEST：

CONFLATED -> {
    require(onBufferOverflow == BufferOverflow.SUSPEND) {
        "CONFLATED capacity cannot be used with non-default onBufferOverflow"
    }
    ConflatedChannel(onUndeliveredElement)
}

Unlimited Channel#

Unlimited Channel 顧名思義，Buffer 大小不設限：

UNLIMITED -> LinkedListChannel(onUndeliveredElement)

它的內部實作 LinkedListChannel 把 isBufferAlwaysFull、isBufferFull 都設成 false，永遠不會滿，因此也不會走到 onBufferOverflow 的邏輯（最終仍會受到記憶體限制）。

class Day16 {
    val scope = CoroutineScope(Job())
    suspend fun conflatedChannel(): Unit {
        val channel = Channel<Int>(capacity = Channel.UNLIMITED)
        scope.launch {
            for (i in 0..10) {
                channel.send(threeTimesInt(i))
            }
        }

        for (i in 0..10) {
            println(channel.receive())
        }
        println("done")
    }

    private suspend fun threeTimesInt(x: Int): Int {
        delay(100L * x)
        return x * 3
    }
}

Unlimited Channel 與 Rendezvous Channel 看起來相似，差別在於 RendezvousChannel 的 isBufferFull 永遠是 true，所以 send 一定會暫停；而 Unlimited 永遠不滿，send 永遠不暫停。

Buffered Channel（自訂容量）#

當 capacity 是一個正整數，Buffer 既不是 0 也不是無限大時：

send 只有在 Buffer 滿的時候才會暫停。
receive 只有在 Buffer 空的時候才會暫停。

從 Channel builder 來看，當 capacity == 1 且 onBufferOverflow == BufferOverflow.DROP_OLDEST 時，會建立 ConflatedChannel；否則會建立 ArrayChannel：

if (capacity == 1 && onBufferOverflow == BufferOverflow.DROP_OLDEST)
    ConflatedChannel(onUndeliveredElement)
else
    ArrayChannel(capacity, onBufferOverflow, onUndeliveredElement)

ArrayChannel 內部會根據傳入的 capacity 與當前狀態判斷 Buffer 是否滿：

internal open class ArrayChannel<E>(
    private val capacity: Int,
    private val onBufferOverflow: BufferOverflow,
    onUndeliveredElement: OnUndeliveredElement<E>?
) {
    protected final override val isBufferEmpty: Boolean get() = state.size == 0
    protected final override val isBufferFull: Boolean get() = state.size == capacity
}

小結#

Channel 大致分為四種，可以依照需求選用。
receive 行為固定：channel 為空時暫停，直到有新元素為止。
send 行為依 Buffer 容量而定：UNLIMITED 永不暫停；RENDEZVOUS 一定暫停；CONFLATED 永遠覆蓋舊值；BUFFERED 視 Buffer 是否已滿。
預設使用 RENDEZVOUS，所以 send 與 receive 必須成對呼叫。

原文出處#

原書/iThome：https://ithelp.ithome.com.tw/articles/10269373