Kotlin 协程 (6/6篇) - 跨协程间通信 Channel

一、概念

1.1 通信模式

根据接收端节点数量以及接收方式，可分为三类。

1.2 扇入扇出

二、组播（通道、队列）

        出现在Flow之前，现在退居幕后职责单一，仅作为协程间通信的并发安全的缓冲队列而存在。

        SendChannel在创建时定义了消费方式外界只能往里发送值，ReceiveChannel在创建时定义了生产方式外界只能从中获取值，Channel继承了它俩既能发送也能接收，根据实际需求暴露不同类型收窄功能。

• 非阻塞：类似于 Java 中的 BlockingQueue 队列，不同的是 put() 和 take() 读取写入数据是阻塞的，而 Channel 中的 send() 和 receive() 是挂起的。
• 同步性：每个值只能被众多订阅者中的一个消费。
• 并发安全：没有检测到 receive() 的话 send() 就会挂起不会发送值（默认模式 RENDEZVOUS）。
• 公平性：在多个协程中发送或接收（多线程竞争）遵循先进先出（FIFO即队列）。

• send() 和 receive()：挂起函数。当接收时，Channel中没有元素，协程将被挂起直到元素可用。当发送时，Channel容量达到阈值，协程将被挂起直到容量可用。
• trySend() 和 tryReceive()：普通函数。从非挂起函数中发送或接收元素，操作是即时的并返回 ChannelResult 对象，包含了有关操作成功或失败的信息。
• close() 和 cancel()：当创建的是 Channel 类型的时候，两个都可以用来关闭，close() 会接收完缓冲区中的值，cancel() 直接关闭。通道使用完一定要关闭，否则程序不会停止造成资源浪费。
• produce() 和 actor() 被定义成协程构建器（因此会在异常、完成、取消时自动关闭Channel），同 launch、async 一样作为 CoroutineScope 的扩展函数。

fun main() = runBlocking {
    private val _channel = Channel()
    val receiveChannel: ReceiveChannel = _channel
    val sendChannel: SendChannel = _channel
    
    _channel.send("Channel发送")
    sendChannel.send("SendChannel发送")
    delay(1000)
    _channel.consumeEach { println("Channel接收：$it") }
    receiveChannel.consumeEach { println("ReceiveChannel接收：$it") }
}

2.1 创建

2.1.1 actor() 创建 SendChannel

创建时在协程构建器 actor() 的 Lambda 中定义了数据的消费方式，返回一个生产者通道 SendChannel，其它协程通过该对象往里发送数据。

fun main() = runBlocking {
    //创建生产者通道
    val send: SendChannel<Int> = actor {
        for (i in channel) println(i)    //定义了消费方式
    }
    //其它协程拿来发送数据
    launch {
        (1..3).forEach { sendChannel.send(it) }
    }
}

2.1.2 produce() 创建 ReceiveChannel

创建时在协程构建器 produce() 的 Lambda 中定义了数据的生产方式，返回一个消费者通道 ReceiveChannel，其它协程通过该对象从中取出数据。

fun main() = runBlocking {
    //创建消费者通道
    val receiveChannel: ReceiveChannel<Int> = produce {
        (1..3).forEach { send(it) }    //定义了生产方式
    }
    //其它协程拿来接收数据
    launch {
        for (i in receiveChannel) println(i)
    }
}

2.1.3 构造创建 Channel

val rendezvousChannel = Channel()    //约会类型
val bufferedChannel = Channel(10)    //指定缓存大小类型
val conflatedChannel = Channel(Channel.CONFLATED)    //混合类型
val unlimitedChannel = Channel(Channel.UNLIMITED)    //无限缓存大小类型

fun main() = runBlocking {
    val channel = Channel<Int>(capacity = Channel.CONFLATED) {
        println("onUndeliveredElement: $it")
    }
    launch {
        (1..3).forEach {
            println("send: $it")
            channel.send(it)
        }
        channel.close()    //不要忘记关闭
    }
    launch {
        for (i in channel) {
            println("receive: $i")
        }
    }
    println("结束！")
}
 
打印：
结束！
send: 1
send: 2
onUndeliveredElement: 1
send: 3
onUndeliveredElement: 2
receive: 3

2.2 遍历元素

2.2.1 迭代器 iterate

fun main(): Unit = runBlocking {
    val channel = Channel(Channel.UNLIMITED)
    repeat(5) { channel.send("$it") }
    val iterator = channel.iterator()
    while (iterator.hasNext()) {
        println("【iterator】${iterator.next()}")
        delay(1000)
    }
    //5个元素打印完后程序没结束，Channel不关闭，后面代码执行不到
    println("这行执行不到")
}

2.2.2 for循环

不会消费掉 Channel 中的值。

fun main():Unit = runBlocking {
    val channel = Channel(Channel.UNLIMITED)
    repeat(5) { channel.send("$it") }
    for (i in channel) {
        println("【for】$i,")
        delay(1000)
    }
    //5个元素打印完后程序没结束，Channel不关闭，后面代码执行不到
    println("这行执行不到")
}

2.2.3 consumeEach( )

会消费掉 Channel 中的值。

fun main(): Unit = runBlocking {
    val channel = Channel(Channel.UNLIMITED)
    repeat(5) { channel.send("$it") }
    channel.consumeEach {
        println("【consumeEach】$it,")
    }
    //5个元素打印完后程序没结束，Channel不关闭，后面代码执行不到
    println("这行执行不到")
}

2.3 转换

Channel 转成 Flow 是为了使用那些方便的操作符，同时 Flow 的很多功能扩展底层由 Channel 实现（如flowOn、buffer）。

三、广播

BroadcastChannel 在协程1.4版本已被废弃，取而代之的是 SharedFlow（StateFlow是它的特定配置版本）。

3.1 SharedFlow

SharedFlow只能订阅（消费），FlowCollector只能发送（生产），MutableSharedFlow继承了它俩既能发送也能订阅，根据实际需求暴露不同类型收窄功能。接收会一直监听，通过取消协程来关闭它的收集。

fun main() = runBlocking {
    //利用多态暴露不同父类限制功能给外部使用
    private val _mutableSharedFlow = MutableSharedFlow()
    val sharedFlow: SharedFlow = _mutableSharedFlow    //或调用 asSharedFlow()
    val flowCollector: FlowCollector = _mutableSharedFlow
 
    launch { _mutableSharedFlow.collect(println("mutableSharedFlow接收：$it")) }
    launch { sharedFlow.collect(println("mutableSharedFlow接收：$it")) }
    delay(1000)
    mutableSharedFlow.emit("mutableSharedFlow发送")
    flowCollector.emit("flowCollector发送")
}

3.1.1 通过构造创建

suspend fun method(): Unit = coroutineScope {   //让3个launch同时进行
    val shared = MutableSharedFlow<Int>(3)    //回放3个数据
    launch {
        for(i in 1..5){
            shared.emit(i)
            println("emmit：$i")
            delay(1000) //1秒更新一个值
        }
    }
    //订阅者甲
    launch { shared.collect{ println("甲：$it") } }
    //订阅者乙5秒后再订阅，也就是值全都更新完了再订阅
    delay(5000)
    launch { shared.collect{ println("乙：$it") } }
}
 
打印：
emmit：1
甲：1
emmit：2
甲：2
emmit：3
甲：3
emmit：4
甲：4
emmit：5
甲：5
乙：3    //回放了3个已更新过的数据
乙：4
乙：5

3.1.2 转换 Flow → SharedFlow

3.2 StateFlow

StateFlow继承自SharedFlow是一种特殊配置，相当于MutableSharedFlow(1,0, BufferOverflow.DROP_OLDEST)，可以使用value属性来访问值，可以当作是用来取代LiveData。

• 必须传入默认值：Null安全。
• 回放个数1+额外缓冲区大小0：只持有1个值。
• 缓存策略DROP_OLDEST：只持有最新值（新值会替换旧值）。
• 数据防抖：即仅在新值内容发生变化才会消费。

3.2.1 通过构造创建

val state = MutableStateFlow(0) //默认值会被覆盖
launch {
    for(i in 1..5){
        state.emit(i)
        println("emit：$i")
        delay(1000) //1秒更新一个值
    }
}
launch {
    delay(2000) //2秒后开始订阅
    state.collect{ println("collect：$it") }
}
 
打印：
emit：1
emit：2
collect：2 //2秒后开始订阅，不会收到之前已更新的数据
emit：3
collect：3
emit：4
collect：4
emit：5
collect：5

3.2.2 转换 Flow → StateFlow

四、区别及选择

4.1 组播(Channel) or 广播(SharedFlow)

4.2 事件(Event) or 状态(State)

事件按顺序都要执行到，状态只关心最新值。