Kotlin语言（十二）：Channel

注：本文中使用 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!
}