注:本文中使用 runBlocking 是为了方便测试,业务开发中禁止使用
一、Channel 基础
(1)Channel 翻译过来为通道或者管道,实际上就是个队列, 是一个面向多协程之间数据传输的 BlockQueue
,用于协程间通信;
(2)Channel 使用 send
和 receive
两个方法往管道里面写入和读取数据,这两个方法是非阻塞的挂起函数;
(3)Channel 是热流,不管有没有订阅者,上游都会发射数据。
1、简单使用
fun channelFun() = runBlocking {
val channel = Channel<Int>()
launch {
for (x in 1..5) channel.send(x * x)
}
repeat(5) {
println(channel.receive())
}
println("Done!")
// 1
// 4
// 9
// 16
// 25
// Done!
}
2、Channel 的迭代
(1)我们发现,这种方式,实际上是我们一直在等待读取 Channel 中的数据,只要有数据到了,就会被读取到;
(2)最后一行 Done! 没有打印出来,表示程序没有结束,一直处于等待读取数据的状态。
fun channelIteratorFun() = runBlocking {
val channel = Channel<Int>()
launch {
for (x in 1..5) {
channel.send(x * x)
}
}
/*val iterator = channel.iterator()
while (iterator.hasNext()) {
println(iterator.next())
}*/
for (y in channel) {
println(y)
}
println("Done")
// 1
// 4
// 9
// 16
// 25
}
3、关闭 Channel
(1)调用 close
方法就像向通道发送了一个特殊的关闭指令,这个迭代停止,说明关闭指令已经被接收了;
(2)这里能够保证所有先前发送出去的元素都能在通道关闭前被接收到;
(3)调用了 close
会立即停止接受新元素,isClosedForSend
会立即返回 true
,而由于 Channel
缓冲区的存在,这时候可能还有一些元素没有被处理完,所以要等所有的元素都被读取之后 isClosedForReceive
才会返回 true
。
fun channelCloseFun() = runBlocking {
val channel = Channel<Int>()
launch {
for (x in 1..5) {
channel.send(x * x)
}
channel.close() //结束发送数据
}
for (y in channel) {
println(y)
}
println("Done!")
// 1
// 4
// 9
// 16
// 25
// Done!
}
4、Channel 的类型
(1)Channel
是一个接口,它继承了 SendChannel
和 ReceiveChannel
两个接口
(2)SendChannel
提供了发射数据的功能,有如下重点接口:
- send
是一个挂起函数,将指定的元素发送到此通道,在该通道的缓冲区已满或不存在时挂起调用者。如果通道已经关闭,调用发送时会抛出异常;
- trySend
如果不违反其容量限制,则立即将指定元素添加到此通道,并返回成功。否则,返回失败或关闭;
- close
关闭通道;
- isClosedForSend
判断通道是否已经关闭,如果关闭,调用 send 会引发异常。
(3)ReceiveChannel
提供了接收数据的功能,有如下重点接口:
- receive
如果此通道不为空,则从中检索并删除元素;如果通道为空,则挂起调用者;如果通道未接收而关闭,则引发 ClosedReceiveChannel 异常;
- tryReceive
如果此通道不为空,则从中检索并删除元素,返回成功结果;如果通道为空,则返回失败结果;如果通道关闭,则返回关闭结果;
- receiveCatching
如果此通道不为空,则从中检索并删除元素,返回成功结果;如果通道为空,则返回失败结果;如果通道关闭,则返回关闭的原因;
- isEmpty
判断通道是否为空;
- isClosedForReceive
判断通道是否已经关闭,如果关闭,调用 receive 会引发异常;
- cancel(cause: CancellationException? = null)
以可选原因取消接收此频道的剩余元素,此函数用于关闭通道并从中删除所有缓冲发送的元素;
- iterator()
返回通道的迭代器。
(4)创建不同类型的 Channel
- Rendezvous channel
0尺寸 buffer (默认类型)
- Unlimited channel
无限元素, send 不被挂起
- Buffered channel
指定大小, 满了之后 send 挂起
- Conflated channel
新元素会覆盖旧元素, receiver 只会得到最新元素, send 永不挂起
fun channelCreateFun() = runBlocking {
val rendezvousChannel = Channel<String>()
val bufferedChannel = Channel<String>(10)
val conflatedChannel = Channel<String>(Channel.CONFLATED)
val unlimitedChannel = Channel<String>(Channel.UNLIMITED)
}
二、Channel实现协程间通信
1、多个协程访问同一个 Channel
fun multipleCoroutineFun() = runBlocking {
val channel = Channel<Int>()
launch {
for (x in 1..3) {
channel.send(x)
}
}
launch {
delay(10)
for (y in channel) {
println("1 --> $y")
}
}
launch {
delay(20)
for (y in channel) {
println("2 --> $y")
}
}
launch {
delay(30)
for (x in 90..93) {
channel.send(x)
}
channel.close()
}
delay(1000)
println("Done!")
// 1 --> 1
// 2 --> 3
// 1 --> 2
// 2 --> 90
// 2 --> 92
// 1 --> 91
// 2 --> 93
// Done!
}
2、produce 和 actor
(1)通过 produce
这个方法启动一个生产者协程,并返回一个 ReceiveChannel
,其他协程就可以拿着这个 Channel 来接收数据了;
(2)通过 actor
可以用来构建一个消费者协程,并返回一个 SendChannel
,其他协程就可以拿着这个 Channel 来发送数据了。
fun produceFun() = runBlocking {
val receiveChannel = produce {
for (x in 1..3) {
delay(500)
send(x)
}
}
for (x in receiveChannel) {
println(x)
}
delay(3000)
receiveChannel.cancel()
println("Done!")
// 1
// 2
// 3
// Done!
}
fun actorFun() = runBlocking {
val sendChannel = actor<Int> {
for (e in channel) {
println(e)
}
}
sendChannel.send(100)
delay(2000)
sendChannel.close()
println("Done!")
// 100
// Done!
}
3、BroadcastChannel
(1)BroadcastChannel
被标记为过时了,请使用 SharedFlow
和 StateFlow
替代它;
(2)1中例子提到一对多的情形,从数据处理本身来讲,有多个接收端的时候,同一个元素只会被一个接收端读到;而 BroadcastChannel
则不然,多个接收端不存在互斥现象。
fun broadcastChannelFun() = runBlocking {
val broadcastChannel = BroadcastChannel<Int>(5)
val receiveChannel1 = broadcastChannel.openSubscription()
val receiveChannel2 = broadcastChannel.openSubscription()
launch {
for (x in 1..3) {
broadcastChannel.send(x)
}
}
launch {
for (e in receiveChannel1) {
println("1 --> $e")
}
}
launch {
for (e in receiveChannel2) {
println("2 --> $e")
}
}
delay(1000)
broadcastChannel.close()
println("Done!")
// 1 --> 1
// 1 --> 2
// 1 --> 3
// 2 --> 1
// 2 --> 2
// 2 --> 3
// Done!
}
使用 broadcast()
扩展函数可以将 Channel
转换成 BroadcastChannel
fun broadcastChannelFun2() = runBlocking {
val channel = Channel<Int>()
val broadcastChannel = channel.broadcast(3)
val receiveChannel1 = broadcastChannel.openSubscription()
val receiveChannel2 = broadcastChannel.openSubscription()
launch {
for (x in 1..3) {
channel.send(x)
}
}
launch {
for (e in receiveChannel1) {
println("1 --> $e")
}
}
launch {
for (e in receiveChannel2) {
println("2 --> $e")
}
}
delay(1000)
channel.close()
println("Done!")
// 1 --> 1
// 1 --> 2
// 1 --> 3
// 2 --> 1
// 2 --> 2
// 2 --> 3
// Done!
}