Kotlin语言(十二):Channel

注:本文中使用 runBlocking 是为了方便测试,业务开发中禁止使用

一、Channel 基础

(1)Channel 翻译过来为通道或者管道,实际上就是个队列, 是一个面向多协程之间数据传输的 BlockQueue,用于协程间通信;
(2)Channel 使用 sendreceive 两个方法往管道里面写入和读取数据,这两个方法是非阻塞的挂起函数;
(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 是一个接口,它继承了 SendChannelReceiveChannel 两个接口

(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 被标记为过时了,请使用 SharedFlowStateFlow 替代它;
(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!
}
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
禁止转载,如需转载请通过简信或评论联系作者。
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 206,214评论 6 481
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 88,307评论 2 382
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 152,543评论 0 341
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 55,221评论 1 279
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 64,224评论 5 371
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,007评论 1 284
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,313评论 3 399
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,956评论 0 259
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,441评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,925评论 2 323
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,018评论 1 333
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,685评论 4 322
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,234评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,240评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,464评论 1 261
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,467评论 2 352
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,762评论 2 345

推荐阅读更多精彩内容