【Kotlin 协程】Channel 通道 ④ ( Channel 通道的热数据流属性 | Channel 通道关闭过程 | Channel 通道关闭代码示例 )

韩曙亮

发布于 2023-03-30 18:32:59

4920

发布于 2023-03-30 18:32:59

文章被收录于专栏：韩曙亮的移动开发专栏

文章目录

一、Channel 通道的热数据流属性

调用 CoroutineScope#produce 函数构造的生产者协程 , 以及调用 CoroutineScope#actor 函数构造的消费者协程 ,

如果上述生产者协程和消费者协程执行完毕 , 则对应的 Channel 通道也会进行关闭 ,

因此 , Channel 通道被称为热数据流 ;

与 Channel 通道热数据流相对的是 Flow 异步流的冷数据流 特征 ;

二、Channel 通道关闭过程

如果调用 Channel 通道的 Channel#close 函数 , 该 Channel 通道会停止接收新的数据 ,

此时调用 Channel#isClosedForSend 函数 , 会返回 true ;

Channel 通道存在缓冲区 , 通道不接收新的元素 , 但是缓冲区已存储的元素需要被处理完毕 , 然后才能关闭通道 ,

当 Channel 通道缓冲区所有的元素处理完毕 , 调用 Channel#isClosedForReceive 函数会返回 true ;

三、Channel 通道关闭代码示例

在下面的代码中 , Channel 通道缓冲区大小为 3 , 数据生产者一次性将 3 个数据全部发送出去 , 但是数据消费者每秒只能消费一个数据 , 需要 3 秒才能将数据处理完毕 ;

在发送完数据后 , 调用 Channel#close 函数 , 关闭通道 , 此时 channel.isClosedForSend 为 true , 但是 channel.isClosedForReceive 为 false ;

在接收数据完毕后 , channel.isClosedForReceive 才变为 true , Channel 通道正式关闭 ;

代码示例 :

package kim.hsl.coroutine

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel

class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        runBlocking {
            // Channel 通道, 传递 Int 类型数据
            val channel = Channel<Int>(3)

            // 数据生产者协程
            val producer = GlobalScope.launch {
                for (i in 0..2) {
                    channel.send(i)
                    println("向通道中发送数据 $i, channel.isClosedForSend : ${channel.isClosedForSend}, channel.isClosedForReceive : ${channel.isClosedForReceive}")
                }
                // 关闭通道
                channel.close()
                println("数据发送完毕, channel.isClosedForSend : ${channel.isClosedForSend}, channel.isClosedForReceive : ${channel.isClosedForReceive}")
            }

            // 数据消费者协程
            val consumer = GlobalScope.launch {
                for(num in channel) {
                    delay(1000)
                    println("从通道中接收数据 $num, channel.isClosedForSend : ${channel.isClosedForSend}, channel.isClosedForReceive : ${channel.isClosedForReceive}")
                }
                println("数据接收完毕, channel.isClosedForSend : ${channel.isClosedForSend}, channel.isClosedForReceive : ${channel.isClosedForReceive}")

            }

            // 等待两个协程执行完毕
            joinAll(producer, consumer)
        }
    }
}

执行结果 :

2022-12-28 11:07:06.509 I/System.out: 向通道中发送数据 0, channel.isClosedForSend : false, channel.isClosedForReceive : false
2022-12-28 11:07:06.509 I/System.out: 向通道中发送数据 1, channel.isClosedForSend : false, channel.isClosedForReceive : false
2022-12-28 11:07:06.509 I/System.out: 向通道中发送数据 2, channel.isClosedForSend : false, channel.isClosedForReceive : false
2022-12-28 11:07:06.510 I/System.out: 数据发送完毕, channel.isClosedForSend : true, channel.isClosedForReceive : false
2022-12-28 11:07:07.516 I/System.out: 从通道中接收数据 0, channel.isClosedForSend : true, channel.isClosedForReceive : false
2022-12-28 11:07:08.527 I/System.out: 从通道中接收数据 1, channel.isClosedForSend : true, channel.isClosedForReceive : false
2022-12-28 11:07:09.540 I/System.out: 从通道中接收数据 2, channel.isClosedForSend : true, channel.isClosedForReceive : true
2022-12-28 11:07:09.540 I/System.out: 数据接收完毕, channel.isClosedForSend : true, channel.isClosedForReceive : true