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

韩曙亮

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

8250

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

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

文章目录

一、Channel 通道

协程中的 Channel 通道是并发的安全队列 , 不同的协程之间可以借助 Channel 通道进行通信 ;

协程中也涉及到生产消费模式 ,

生产者协程产生数据 , 将数据通过 Channel 通道 , 发送给消费者协程消费数据 ;

Flow 流是通过发送元素 , 然后再收集元素 , Flow 流是冷流 , 其发送元素需要通过收集元素触发 , 也就是如果不收集元素 , 发送元素的操作是不执行的 ;

Channel 通道与 Flow 异步流是不同的 , 生产者协程产生数据和消费者协程消费数据是同时进行的 ;

二、Channel#send 发送数据

Channel#send 函数原型 :

/**
 * 将指定的[element]发送到此通道，在此通道的缓冲区已满时挂起调用者
 * 或者如果它不存在，或者如果通道[is closed for `send`][isClosedForSend]抛出异常(详情参见[close])s).
 *
 * [Closing][close]通道_after_此函数已挂起不会导致此挂起[send]调用
 * 因为关闭一个通道在概念上就像在这个通道上发送一个特殊的“关闭令牌”。
 * 通过通道发送的所有元素都按照先进先出的顺序交付。发送元素
 * 将在关闭令牌之前交付给接收者。
 *
 * 这个暂停函数是可以取消的。如果当前协程的[Job]被取消或完成
 * 函数挂起后，该函数立即恢复并返回[CancellationException]。
 * 有**立即取消保证**。如果作业被取消，而此函数被取消
 * 暂停，将无法成功恢复。' send '调用可以将元素发送到通道，
 * 但随后抛出[CancellationException]，因此异常不应被视为交付元素失败。
 * 有关处理未交付元素的详细信息，请参见[Channel]文档中的“未交付元素”部分。
 *
 * 注意，该函数在未挂起时不会检查是否取消。
 * 使用[yield]或[CoroutineScope.isActive]如果需要，在紧循环中定期检查取消。
 *
 * 这个函数可以在[select]调用中使用[onSend]子句。
 * 使用[trySend]尝试发送到此通道而不等待。
 */
public suspend fun send(element: E)

三、Channel#receive 接收数据

Channel#receive 函数原型 :

/**
 * 检索并删除该通道中不为空的元素，或在通道为空时挂起调用者，
 * 或者如果通道[为' receive '关闭][isClosedForReceive]则抛出[ClosedReceiveChannelException]。
 * 如果通道由于异常而关闭，则称为_failed_通道，此函数
 * 将抛出原始的[close][SendChannel。关闭]导致异常。
 *
 * 这个暂停函数是可以取消的。如果当前协程的[Job]被取消或完成
 * 函数挂起后，该函数立即恢复并返回[CancellationException]。
 * 有**立即取消保证**。如果作业被取消，而此函数被取消
 * 暂停，将无法成功恢复。' receive '调用可以从通道中检索元素，
 * 但随后抛出[CancellationException]，从而无法交付元素。
 * 有关处理未交付元素的详细信息，请参见[Channel]文档中的“未交付元素”部分。
 *
 * 注意，该函数在未挂起时不会检查是否取消。
 * 使用[yield]或[CoroutineScope。isActive]如果需要，在紧循环中定期检查取消。
 *
 * 这个函数可以在[select]调用中使用[onReceive]子句。
 * 使用[tryReceive]尝试从此通道接收而不等待。
 */
public suspend fun receive(): E

四、Channel 通道代码示例

代码示例 :

package kim.hsl.coroutine

import android.os.Bundle
import androidx.appcompat.app.AppCompatActivity
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

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

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

            // 数据生产者协程
            val producer = GlobalScope.launch {
                for (i in 0..3) {
                    delay(1000)
                    channel.send(i)
                    println("向通道中发送数据 $i")
                }
            }

            // 数据消费者协程
            val consumer = GlobalScope.launch {
                while (true) {
                    val num = channel.receive()
                    println("从通道中获取数据 $num")
                }
            }
        }
    }
}

执行结果 :

2022-12-27 11:54:15.689 I/System.out: 向通道中发送数据 0
2022-12-27 11:54:15.690 I/System.out: 从通道中获取数据 0
2022-12-27 11:54:16.707 I/System.out: 向通道中发送数据 1
2022-12-27 11:54:16.708 I/System.out: 从通道中获取数据 1
2022-12-27 11:54:17.745 I/System.out: 向通道中发送数据 2
2022-12-27 11:54:17.746 I/System.out: 从通道中获取数据 2
2022-12-27 11:54:18.749 I/System.out: 向通道中发送数据 3
2022-12-27 11:54:18.749 I/System.out: 从通道中获取数据 3