Kotlin协程-Channel基础

Posted 且听真言

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了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)

以上是关于Kotlin协程-Channel基础的主要内容，如果未能解决你的问题，请参考以下文章

Kotlin 协程Channel 通道 ① ( Channel#send 发送数据 | Channel#receive 接收数据 )

Kotlin 协程Channel 通道 ① ( Channel#send 发送数据 | Channel#receive 接收数据 )

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

深潜Kotlin协程（十六）：Channel

深潜Kotlin协程（十六）：Channel

Kotlin 协程Channel 通道 ⑤ ( BroadcastChannel 广播通道 | 代码示例 )