Day15：Channel 的第一堂課

13th鐵人賽 kotlin讀書會 kotlin coroutine

Andy 安迪

團隊Kotlin 愛台灣 2021

2021-09-20 23:01:03

857 瀏覽

分享至

在前面的文章中，我們介紹了 coroutine 的基本原理，如何使用 launch 、 async 建立 CoroutineScope，如何選擇適當的調度器，讓任務跑在適合的執行緒/執行緒池中等等... 不過任務在 launch 、 async 建立的範圍中，必須要等到任務執行完成之後才能夠回傳，假設我們希望能夠在執行到一半的時候就把目前的值傳出來，我們該怎麼做呢？

換句話說，如果在 launch、async 中需要同時執行好幾項任務，但是我們並不需要等待所有任務完成之後才回傳，我們希望能夠在一項任務完成之後就立刻回傳。

Kotlin Coroutine 提供了 Channel 可以用來解決這個問題。

Channel 的意思是「通道」，也就是說我們可以在不同的 CoroutineScope 建立一個通道，讓資料可以由這個通道傳至其他的 Scope。

WHOOSHH’S INNOVATIVE FISH PASSAGE SOLUTION

上圖是某公司替鮭魚設計的一個洄游的解決方案XD

Channel Basic

讓我們建立第一個 Channel ：

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

    suspend fun firstChannel() {
        val channel = Channel<Int>()
        scope.launch {
            repeat(10) {
                delay(200)
                channel.send(it)
            }
        }
        repeat(10) {
            println(channel.receive())
        }
        println("Finish!")
    }
}

在這個 suspend 函式中，我們假設有一項任務需要做十次，每一次都需要執行 200 豪秒，我們需要在每一次的結果產生時都先將這個值傳出去給其他 CoroutineScope 使用。

First Channel

從上方範例我們發現， Channel 可以在不同的 CoroutineScope 中傳送資料，如上方的 scope.launch → outer scope。

FIFO

Channel 的機制類似 Queue，資料是採取FIFO(First in first out)的方式傳出。

在 Channel 中，我們使用 send() 來把資料傳進 Channel 中，而使用 receive() 將資料取出。

Kotlin Channel

send() 以及 receive()

如果我們將 receive() 呼叫的次數增加一次，如下：

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

    suspend fun firstChannel() {
        val channel = Channel<Int>()
        scope.launch {
            repeat(10) {
                delay(200)
                channel.send(it)
            }
        }
        repeat(11) {
            println(channel.receive())
        }
        println("Finish!")
    }
}

結果會是

Imgur

coroutine 不會結束， receive() 會持續等待 channel 提供資料讓它取出。為什麼 receive() 可以等待呢？我想你可能猜到了，因為 receive() 也是一個 suspend 函式。

public suspend fun receive(): E

receive()

receive() 的動作是如果 channel 裏面不為空，那麼就會取出資料，否則將會暫停呼叫端的 coroutine。所以上面的範例如果呼叫 receive() 的次數比 send() 的次數還要多，那麼就會暫停 coroutine，直到 channel 又有值可以讓 recieve() 取出。

send()

其實， send() 也是一個 suspend 函式，只不過預設是沒有作用的。

public suspend fun send(element: E)

使用 Channel() 建立 channel 的時候，如果沒有帶任何參數，那麼建立出來的 channel 是屬於沒有緩衝區 (Buffer) 的 Rendezvous channel。

使用 Rendezvous channel 來傳輸資料，因為沒有 Buffer ，所以傳輸資料的方式會是當有 send() 以及 receive() 都被呼叫的時候，這個時候資料才會傳輸過去，這種方式就不需要 Buffer ，呼叫 send() 時也就不需要暫停了。

小結

Channel 是採取 FIFO 的資料傳輸方式，也就是先傳進去的資料會先取出來。而在 channel 中，如果使用 Channel() 建立 channel 時沒有帶入任何參數，那麼預設的容量(Capacity)就是為 0，而這邊的容量指的是緩衝區(Buffer)的容量，當沒有緩衝區時，資料要經由 channel 傳輸，則必須有人呼叫 send() 以及 receive()，這兩個的呼叫的次數必須要一樣，如果 receive() 呼叫的次數比 send() 還要多，表示 receive() 從 channel 取出資料的時候有可能會取不到資料，這個時候 receive() 就會暫停目前的 coroutine ，直到有另外的資料傳入這個 channel()。