Kotlin语言（十二）：Channel

Posted 2023-03-25

参考技术A

（1） Channel 翻译过来为通道或者管道，实际上就是个队列, 是一个面向多协程之间数据传输的 BlockQueue ，用于协程间通信；
（2） Channel 使用 send 和 receive 两个方法往管道里面写入和读取数据，这两个方法是非阻塞的挂起函数；
（3） Channel 是热流，不管有没有订阅者，上游都会发射数据。

（1）我们发现，这种方式，实际上是我们一直在等待读取 Channel 中的数据，只要有数据到了，就会被读取到；
（2）最后一行 Done! 没有打印出来，表示程序没有结束，一直处于等待读取数据的状态。

（1）调用 close 方法就像向通道发送了一个特殊的关闭指令，这个迭代停止，说明关闭指令已经被接收了；
（2）这里能够保证所有先前发送出去的元素都能在通道关闭前被接收到；
（3）调用了 close 会立即停止接受新元素， isClosedForSend 会立即返回 true ，而由于 Channel 缓冲区的存在，这时候可能还有一些元素没有被处理完，所以要等所有的元素都被读取之后 isClosedForReceive 才会返回 true 。

（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 永不挂起

（1）通过 produce 这个方法启动一个生产者协程，并返回一个 ReceiveChannel ，其他协程就可以拿着这个 Channel 来接收数据了；
（2）通过 actor 可以用来构建一个消费者协程，并返回一个 SendChannel ，其他协程就可以拿着这个 Channel 来发送数据了。

（1） BroadcastChannel 被标记为过时了，请使用 SharedFlow 和 StateFlow 替代它；
（2）1中例子提到一对多的情形，从数据处理本身来讲，有多个接收端的时候，同一个元素只会被一个接收端读到；而 BroadcastChannel 则不然，多个接收端不存在互斥现象。

使用 broadcast() 扩展函数可以将 Channel 转换成 BroadcastChannel

Kotlin协程-Channel基础

一、Channel——管道

1.创建Channel

Channel 是一个管道。Channel分为接收方和发送方，发送方发送数据，接收方接收数据。

fun main() 
    runBlocking 
        val channel = Channel<Int>()

        launch 
            (1..3).forEach 
                channel.send(it)
                printChannelCoroutine("Channel Send:$it")
            
        

        launch 
            for (i in channel) 
                printChannelCoroutine("Channel Receive:$i")
            
        

        printCoroutine("End!")
    



Log:
End!;Thread:main @coroutine#1
Channel Receive:1;Thread:main @coroutine#3
Channel Send:1;Thread:main @coroutine#2
Channel Send:2;Thread:main @coroutine#2
Channel Receive:2;Thread:main @coroutine#3
Channel Receive:3;Thread:main @coroutine#3
Channel Send:3;Thread:main @coroutine#2

通过Log可以发现@coroutine#3接收、@coroutine#2发送。Channel可以跨越不同的协程进行通信。在 @coroutine#1当中创建的 Channel，然后分别在 @coroutine#2、@coroutine#3 当中使用 Channel 来传递数据。

可以发现：上面的例子中通过泛型设置Channel中传递的数据类型是Int，又在协程中调用send方法，把数据发送到管道里。send是一个挂起函数。通过for循环遍历Channel，取出数据。最重要的是程序在输出完所有的结果后，程序仍处于运行状态。

2.关闭Channel

可以通过调用close方法关闭Channel，Channel需要用户手动关闭。

runBlocking 
        val channel = Channel<Int>()

        launch 
            (1..3).forEach 
                channel.send(it)
                printChannelCoroutine("Channel Send:$it")
            

            channel.close()
        

        launch 
            for (i in channel) 
                printChannelCoroutine("Channel Receive:$i")
            
        
        printCoroutine("End!")
    

Log:
End!;Thread:main @coroutine#1
Channel Receive:1;Thread:main @coroutine#3
Channel Send:1;Thread:main @coroutine#2
Channel Send:2;Thread:main @coroutine#2
Channel Receive:2;Thread:main @coroutine#3
Channel Receive:3;Thread:main @coroutine#3
Channel Send:3;Thread:main @coroutine#2

3.Channel源码

public fun <E> Channel(
    capacity: Int = RENDEZVOUS,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    onUndeliveredElement: ((E) -> Unit)? = null
): Channel<E> 

可以发现Channel是一个普通函数，带有一个泛型参数E，有三个参数：

• capacity：指管道容量。默认值为:RENDEZVOUS，此时Channel的容量为0。

  public const val RENDEZVOUS: Int = 0

 capacity的其他取值：

1.无限容量。

public const val UNLIMITED: Int = Int.MAX_VALUE

2.容量为 1，老数据会被新数据替代。

   public const val CONFLATED: Int = -1

3.具备一定的缓存容量，默认情况下是 64，具体容量由这个 VM 参数决定 "kotlinx.coroutines.channels.defaultBuffer"

public const val BUFFERED: Int = -2

• onBufferOverflow：用来处理当管道的容量满时，Channel的策略。

public enum class BufferOverflow 
    /**
     * Suspend on buffer overflow.
     */
    SUSPEND,

    /**
     * Drop **the oldest** value in the buffer on overflow, add the new value to the buffer, do not suspend.
     */
    DROP_OLDEST,

    /**
     * Drop **the latest** value that is being added to the buffer right now on buffer overflow
     * (so that buffer contents stay the same), do not suspend.
     */
    DROP_LATEST

1.SUSPEND,当管道容量已满，发送方仍然发送数据，就会挂起当前send方法。（发送流程就被挂起了），等管道中有空闲位置后再恢复。

2.DROP_OLDEST：丢弃最旧的数据，发送方发送新的数据。

3.DROP_LATEST：丢弃最新的数据（即将发送的数据），管道中内容保持不变。

• onUndeliveredElement：异常处理回调，当管道中某些数据没有被成功接收的时候，这个回调就会被调用。

4.capacity使用

• capacity = UNLIMITED

  runBlocking 
        val channel = Channel<Int>(capacity = Channel.Factory.UNLIMITED)
        launch 
            (1..3).forEach 
                channel.send(it)
                printChannelCoroutine("Channel Send:$it")
            
            channel.close()
        

        launch 
            for (i in channel) 
                printChannelCoroutine("Channel Receive:$i")
            
        
        printCoroutine("End!")
    

Log:
End!;Thread:main @coroutine#1
Channel Send:1;Thread:main @coroutine#2
Channel Send:2;Thread:main @coroutine#2
Channel Send:3;Thread:main @coroutine#2
Channel Receive:1;Thread:main @coroutine#3
Channel Receive:2;Thread:main @coroutine#3
Channel Receive:3;Thread:main @coroutine#3

 从上面的代码可以发现，当在创建Channel的时候,设置了 capacity = Channel.Factory.UNLIMITED。对于发送方来说，由于 Channel 的容量是无限大的，所以发送方可以一直往管道当中塞入数据，等数据都塞完以后，接收方才开始接收。

•  capacity = CONFLATED

   runBlocking 
        val channel = Channel<Int>(capacity = Channel.Factory.CONFLATED)
        launch 
            (1..3).forEach 
                channel.send(it)
                printChannelCoroutine("Channel Send:$it")
            
            channel.close()
        

        launch 
            for (i in channel) 
                printChannelCoroutine("Channel Receive:$i")
            
        
    

Log:
Channel Send:1;Thread:main @coroutine#2
Channel Send:2;Thread:main @coroutine#2
Channel Send:3;Thread:main @coroutine#2
Channel Receive:3;Thread:main @coroutine#3

当设置 capacity = CONFLATED 的时候，发送方也会一直发送数据，而且，对于接收方来说，它永远只能接收到最后一条数据。

5.onBufferOverflow使用

• 运用 onBufferOverflow 与 capacity，来实现 CONFLATED 的效果

 runBlocking 
        val channel = Channel<Int>(capacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
        launch 
            (1..3).forEach 
                channel.send(it)
                printChannelCoroutine("Channel Send:$it")
            
            channel.close()
        

        launch 
            for (i in channel) 
                printChannelCoroutine("Channel Receive:$i")
            
        
        printCoroutine("End!")
    

Log:
End!;Thread:main @coroutine#1
Channel Send:1;Thread:main @coroutine#2
Channel Send:2;Thread:main @coroutine#2
Channel Send:3;Thread:main @coroutine#2
Channel Receive:3;Thread:main @coroutine#3

其实 capacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST，就代表了 capacity = CONFLATED。

• onBufferOverflow = BufferOverflow.DROP_LATEST

        runBlocking 
        val channel = Channel<Int>(capacity = 3, onBufferOverflow = BufferOverflow.DROP_LATEST)
        launch 
            (1..3).forEach 
                channel.send(it)
                printChannelCoroutine("Channel Send:$it")
            
            channel.send(4)
            printChannelCoroutine("Channel Send:4")
            channel.send(5)
            printChannelCoroutine("Channel Send:5")
            channel.close()
        

        launch 
            for (i in channel) 
                printChannelCoroutine("Channel Receive:$i")
            
        
        printCoroutine("End!")
    

Log:
End!;Thread:main @coroutine#1
Channel Send:1;Thread:main @coroutine#2
Channel Send:2;Thread:main @coroutine#2
Channel Send:3;Thread:main @coroutine#2
Channel Send:4;Thread:main @coroutine#2
Channel Send:5;Thread:main @coroutine#2
Channel Receive:1;Thread:main @coroutine#3
Channel Receive:2;Thread:main @coroutine#3
Channel Receive:3;Thread:main @coroutine#3

onBufferOverflow = BufferOverflow.DROP_LATEST 就意味着，当 Channel 容量满了以后，之后再继续发送的内容，就会直接被丢弃。

6.onUndeliveredElement

onUndeliveredElement 这个参数的作用

runBlocking 
        val channel = Channel<Int>(Channel.UNLIMITED) 
            printChannelCoroutine("onUndeliveredElement =$it")
        
        (1..3).forEach 
            channel.send(it)
        
        channel.receive()
        channel.cancel()
    

Log:
onUndeliveredElement =2;Thread:main @coroutine#1
onUndeliveredElement =3;Thread:main @coroutine#1

 可以发现，onUndeliveredElement 就是一个回调，当我们发送出去的 Channel 数据无法被接收方处理的时候，就可以通过 onUndeliveredElement 这个回调，来进行监听。

它的使用场景一般都是用于“接收方对数据是否被消费特别关心的场景”。比如说，我发送出去的消息，接收方是不是真的收到了？对于接收方没收到的信息，发送方就可以灵活处理了，比如针对这些没收到的消息，发送方可以先记录下来，等下次重新发送。

7.通过produce 创建Channel

  runBlocking 
        val channel = produce<Int> 
            (1..3).forEach 
                send(it)
                printChannelCoroutine("Send:$it")
            
        

        launch 
            for (i in channel) 
                printCoroutine("Receive:$i")
            
        

        printCoroutine("end")
    

Log:
end;Thread:main @coroutine#1
Receive:1;Thread:main @coroutine#3
Send:1;Thread:main @coroutine#2
Send:2;Thread:main @coroutine#2
Receive:2;Thread:main @coroutine#3
Receive:3;Thread:main @coroutine#3
Send:3;Thread:main @coroutine#2

使用 produce 以后，就不用再去调用 close() 方法了，因为 produce 会自动帮我们去调用 close() 方法。

记录一个异常场景

 runBlocking 
        val channel = produce<Int> 
            (1..3).forEach 
                send(it)
            
        

        channel.receive()
        channel.receive()
        channel.receive()
        channel.receive()
        printCoroutine("End!")
    


Exception in thread "main" kotlinx.coroutines.channels.ClosedReceiveChannelException: Channel was closed
	at kotlinx.coroutines.channels.Closed.getReceiveException(AbstractChannel.kt:1108)
	at kotlinx.coroutines.channels.AbstractChannel$ReceiveElement.resumeReceiveClosed(AbstractChannel.kt:913)
	at kotlinx.coroutines.channels.AbstractChannel.receiveSuspend(AbstractChannel.kt:609)
	at kotlinx.coroutines.channels.AbstractChannel.receive(AbstractChannel.kt:593)
	at kotlinx.coroutines.channels.ChannelCoroutine.receive(ChannelCoroutine.kt)
	at com.example.myapplication.testcoroutinue.TestChannelKt$testChannel9$1.invokeSuspend(TestChannel.kt:34)
	(Coroutine boundary)
	at com.example.myapplication.testcoroutinue.TestChannelKt$testChannel9$1.invokeSuspend(TestChannel.kt:34)
Caused by: kotlinx.coroutines.channels.ClosedReceiveChannelException: Channel was closed

在上面代码中，我们只调用了 3 次 send()，却调用 4 次 receive()。当我们第 4 次调用 receive() 的时候，代码会抛出异常“ClosedReceiveChannelException”，这其实也代表：我们的 Channel 已经被关闭了。所以这也就说明了，produce 确实会帮我们调用 close() 方法。不然的话，第 4 次 receive() 会被挂起，而不是抛出异常。

 runBlocking 
        val channel = Channel<Int>()
        launch 
            (1..3).forEach 
                channel.send(it)
            
        

        channel.receive()
        printCoroutine("Receive: 1")
        channel.receive()
        printCoroutine("Receive: 2")
        channel.receive()
        printCoroutine("Receive: 3")

        channel.receive()
        printCoroutine("End！")
    

Log:
Receive: 1;Thread:main @coroutine#1
Receive: 2;Thread:main @coroutine#1
Receive: 3;Thread:main @coroutine#1

 ，第 4 次调用 receive()，就会导致程序被永久挂起，后面的printCoroutine("End！")

是没有机会继续执行的。也就是说，我们直接使用 receive() 是很容易出问题的。这也是我在前面的代码中一直使用 for 循环，而没有用 receive() 的原因。

8.Channel 还有两个属性：isClosedForReceive、isClosedForSend。

这两个属性，就可以用来判断当前的 Channel 是否已经被关闭。由于 Channel 分为发送方和接收方，所以这两个参数也是针对这两者的。也就是说，对于发送方，我们可以使用“isClosedForSend”来判断当前的 Channel 是否关闭；对于接收方来说，我们可以用“isClosedForReceive”来判断当前的 Channel 是否关闭。

    runBlocking 
        val channel = produce<Int> 
            (1..3).forEach 
                send(it)
                printChannelCoroutine("Send $it")
            
        

        while (!channel.isClosedForReceive) 
            val receive = channel.receive()
            printCoroutine("Receive $receive")
        

        printCoroutine("End")
    

Log:

Send 1;Thread:main @coroutine#2
Receive 1;Thread:main @coroutine#1
Receive 2;Thread:main @coroutine#1
Send 2;Thread:main @coroutine#2
Send 3;Thread:main @coroutine#2
Receive 3;Thread:main @coroutine#1
End;Thread:main @coroutine#1

因为，当你为管道指定了 capacity 以后，以上的判断方式将会变得不可靠！原因是目前的 1.6.0 版本的协程库，运行这样的代码会崩溃，如下所示：

 runBlocking 
        val channel = produce<Int>(capacity = 3) 
            (1..300).forEach 
                send(it)
                printChannelCoroutine("Send $it")
            
        

        while (!channel.isClosedForReceive) 
            val receive = channel.receive()
            printCoroutine("Receive $receive")
        

        printCoroutine("End")
    

Log:
...
...

Receive 299;Thread:main @coroutine#1
Receive 300;Thread:main @coroutine#1
Send 300;Thread:main @coroutine#2
Exception in thread "main" kotlinx.coroutines.channels.ClosedReceiveChannelException: Channel was closed
	at kotlinx.coroutines.channels.Closed.getReceiveException(AbstractChannel.kt:1108)
	at kotlinx.coroutines.channels.AbstractChannel$ReceiveElement.resumeReceiveClosed(AbstractChannel.kt:913)
	at kotlinx.coroutines.channels.AbstractSendChannel.helpClose(AbstractChannel.kt:342)
	at kotlinx.coroutines.channels.AbstractSendChannel.close(AbstractChannel.kt:271)
	at kotlinx.coroutines.channels.SendChannel$DefaultImpls.close$default(Channel.kt:93)
	at kotlinx.coroutines.channels.ProducerCoroutine.onCompleted(Produce.kt:143)
	at kotlinx.coroutines.channels.ProducerCoroutine.onCompleted(Produce.kt:136)
	at kotlinx.coroutines.AbstractCoroutine.onCompletionInternal(AbstractCoroutine.kt:93)
	at kotlinx.coroutines.JobSupport.tryFinalizeSimpleState(JobSupport.kt:294)
	at kotlinx.coroutines.JobSupport.tryMakeCompleting(JobSupport.kt:856)
	at kotlinx.coroutines.JobSupport.makeCompletingOnce$kotlinx_coroutines_core(JobSupport.kt:828)
	at kotlinx.coroutines.AbstractCoroutine.resumeWith(AbstractCoroutine.kt:100)
	(Coroutine boundary)

所以，最好不要用 channel.receive()。即使配合 isClosedForReceive 这个判断条件，我们直接调用 channel.receive() 仍然是一件非常危险的事情！

runBlocking 
        val channel = produce<Int>(capacity = 3) 
            (1..300).forEach 
                send(it)
                printChannelCoroutine("Send $it")
            
        
        channel.consumeEach 
            printChannelCoroutine("Receive $it")
        

        printCoroutine("End")
    

Log:
...
Send 293;Thread:main @coroutine#2
Send 294;Thread:main @coroutine#2
Receive 291;Thread:main @coroutine#1
Receive 292;Thread:main @coroutine#1
Receive 293;Thread:main @coroutine#1
Receive 294;Thread:main @coroutine#1
Receive 295;Thread:main @coroutine#1
Send 295;Thread:main @coroutine#2
Send 296;Thread:main @coroutine#2
Send 297;Thread:main @coroutine#2
Send 298;Thread:main @coroutine#2
Send 299;Thread:main @coroutine#2
Receive 296;Thread:main @coroutine#1
Receive 297;Thread:main @coroutine#1
Receive 298;Thread:main @coroutine#1
Receive 299;Thread:main @coroutine#1
Receive 300;Thread:main @coroutine#1
Send 300;Thread:main @coroutine#2
End;Thread:main @coroutine#1

 所以，当我们想要读取 Channel 当中的数据时，我们一定要使用 for 循环，或者是 channel.consumeEach ，千万不要直接调用 channel.receive()。在某些特殊场景下，如果我们必须要自己来调用 channel.receive()，那么可以考虑使用 receiveCatching()，它可以防止异常发生。

二、为什么说 Channel 是“热”的？

Channel 其实就是用来传递“数据流”的。注意，这里的数据流，指的是多个数据组合形成的流。

在业界一直有一种说法：Channel 是“热”的。也是因为这句话，在 Kotlin 当中，我们也经常把 Channel 称为“热数据流”。

  runBlocking 
        val channel = produce<Int>(capacity = 10) 
            (1..3).forEach 
                send(it)
                printChannelCoroutine("Send $it")
            
        

        printCoroutine("End")
    

Log:
End;Thread:main @coroutine#1
Send 1;Thread:main @coroutine#2
Send 2;Thread:main @coroutine#2
Send 3;Thread:main @coroutine#2

在上面的代码中，我们定义了一个 Channel，管道的容量是 10，然后我们发送了 3 个数据。但你是否注意到了，在代码中并没有消费 Channel 当中的数据。所以，这种“不管有没有接收方，发送方都会工作”的模式，就是我们将其认定为“热”的原因。

是不是因为前面的代码中，设置了“capacity = 10”的原因？如果设置成“capacity = 0”，那 Channel 的发送方是不是就不会主动工作了？让我们来试试。

  runBlocking 
        val channel = produce<Int>(capacity = 0) 
            (1..3).forEach 
                printChannelCoroutine("before Send $it")
                send(it)
                printChannelCoroutine("Send $it")
            
        
        printCoroutine("End !")
    


Log:
End !;Thread:main @coroutine#1
before Send 1;Thread:main @coroutine#2

当我们把 capacity 改成 0 以后，可以看到 Channel 的发送方仍然是会工作的，只是说，在它调用 send() 方法的时候，由于接收方还未就绪，且管道容量为 0，所以它会被挂起。所以，它仍然还是有在工作的。

三、Channel 源码

public interface Channel<E> : SendChannel<E>, ReceiveChannel<E> 

 Channel 本身并没有什么方法和属性，它其实只是 SendChannel、ReceiveChannel 这两个接口的组合。也就是说，Channel 的所有能力，都是来自于 SendChannel、ReceiveChannel 这两个接口。

public interface SendChannel<in E> 
    public val isClosedForSend: Boolean

    public suspend fun send(element: E)

    // 1，select相关
    public val onSend: SelectClause2<E, SendChannel<E>>

    // 2，非挂起函数的接收
    public fun trySend(element: E): ChannelResult<Unit>

    public fun close(cause: Throwable? = null): Boolean

    public fun invokeOnClose(handler: (cause: Throwable?) -> Unit)



public interface ReceiveChannel<out E> 

    public val isClosedForReceive: Boolean

    public val isEmpty: Boolean

    public suspend fun receive(): E

    public suspend fun receiveCatching(): ChannelResult<E>
    // 3，select相关
    public val onReceive: SelectClause1<E>
    // 4，select相关
    public val onReceiveCatching: SelectClause1<ChannelResult<E>>

    // 5，非挂起函数的接收
    public fun tryReceive(): ChannelResult<E>

    public operator fun iterator(): ChannelIterator<E>

    public fun cancel(cause: CancellationException? = null)

 

所以，如果说 Channel 是一个管道，那么 SendChannel、ReceiveChannel 就是组成这个管道的两个零件。

class CHannelModel 

    val channel: ReceiveChannel<Int> by ::_channel

    private val _channel: Channel<Int> = Channel()


    suspend fun init() 
        (1..2).forEach 
            _channel.send(it)
        
    

    fun main() = runBlocking 
        val model = CHannelModel()
        launch 
            model.init()
        

        model.channel.consumeEach 
            println(it)