Coroutines Channels
Java中的多線程通信, 總會涉及到共享狀態(shared mutable state)的讀寫, 有同步, 死鎖等問題要處理.
協程中的Channel用於協程間的通信, 它的宗旨是:
Do not communicate by sharing memory; instead, share memory by communicating.
本文被收錄在: https://github.com/mengdd/KotlinTutorials
Channel basics
channels用於協程間的通信, 允許我們在不同的協程間傳遞數據(a stream of values).
生產者-消費者模式
發送數據到channel的協程被稱為producer, 從channel接受數據的協程被稱為consumer.
生產: send, produce.
消費: receive, consume.
當需要的時候, 多個協程可以向同一個channel發送數據, 一個channel的數據也可以被多個協程接收.
當多個協程從同一個channel接收數據的時候, 每個元素僅被其中一個consumer消費一次. 處理元素會自動將其從channel里刪除.
Channel的特點
Channel在概念上有點類似於BlockingQueue, 元素從一端被加入, 從另一端被消費. 關鍵的區別在於, 讀寫的方法不是blocking的, 而是suspending的.
在為空或為滿時. channel可以suspend它的send和receive操作.
Channel的關閉和迭代
Channel可以被關閉, 說明沒有更多的元素了.
取消producer協程也會關閉channel.
在receiver端有一種方便的方式來接收: 用for迭代.
看這個例子:
fun main() = runBlocking<Unit> {
val channel = Channel<Int>()
launch {
for (x in 1..5) channel.send(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!")
}
運行后會輸出:
1
2
3
4
5
Done!
Process finished with exit code 0
如果注釋掉channel.close()就會變成:
1
2
3
4
5
Done沒有被輸出, 程序也沒有退出, 這是因為接受者協程還在一直等待.
不同的Channel類型
庫中定義了多個channel類型, 它們的主要區別在於:
- 內部可以存儲的元素數量;
send是否可以被掛起.
所有channel類型的receive方法都是同樣的行為: 如果channel不為空, 接收一個元素, 否則掛起.
Channel的不同類型:
- Rendezvous channel: 0尺寸buffer,
send和receive要meet on time, 否則掛起. (默認類型). - Unlimited channel: 無限元素,
send不被掛起. - Buffered channel: 指定大小, 滿了之后
send掛起. - Conflated channel: 新元素會覆蓋舊元素, receiver只會得到最新元素,
send永不掛起.
創建channel:
val rendezvousChannel = Channel<String>()
val bufferedChannel = Channel<String>(10)
val conflatedChannel = Channel<String>(CONFLATED)
val unlimitedChannel = Channel<String>(UNLIMITED)
默認是Rendezvous channel.
練習: 分析代碼輸出
看這段代碼:
fun main() = runBlocking<Unit> {
val channel = Channel<String>()
launch {
channel.send("A1")
channel.send("A2")
log("A done")
}
launch {
channel.send("B1")
log("B done")
}
launch {
repeat(3) {
val x = channel.receive()
log(x)
}
}
}
fun log(message: Any?) {
println("[${Thread.currentThread().name}] $message")
}
這段代碼創建了一個channel, 傳遞String類型的元素.
兩個producder協程, 分別向channel發送不同的字符串, 發送完畢后打印各自的"done".
一個receiver協程, 接收channel中的3個元素並打印.
程序的運行輸出結果會是怎樣呢?
記得在Configurations中加上VM options: -Dkotlinx.coroutines.debug. 可以看到協程信息.
答案揭曉:
[main @coroutine#4] A1
[main @coroutine#4] B1
[main @coroutine#2] A done
[main @coroutine#3] B done
[main @coroutine#4] A2
答對了嗎?
為什么會是這樣呢? 原因主要有兩點:
- 這里創建的channel是默認的Rendezvous類型, 沒有buffer, send和receive必須要meet, 否則掛起.
- 兩個producer和receiver協程都運行在同一個線程上, ready to be resumed也只是加入了一個等待隊列, resume要按順序來.
這個例子在Introduction to Coroutines and Channels中有一個視頻解說.
另外, 官方文檔中還有一個ping-pang的例子, 為了說明Channels are fair.
參考
- 官方文檔: Channels
- Introduction to Coroutines and Channels
- Github: Coroutines Guide
- Kotlin: Diving in to Coroutines and Channels
歡迎關注微信公眾號: 聖騎士Wind
