学校网站建设 分工,vultr做网站,中国建筑网官网查询报考,做软件app需要多少钱前言
在之前的文章中#xff0c;我们已经知道了协程的启动、挂起、取消、异常以及常用的协程作用域等基础应用。 这些基础应用适合的场景是一次性任务#xff0c;执行完就结束了的场景。
launch / async 适合的场景
网络请求数据库查询文件读写并行计算任务等等
Channel …前言
在之前的文章中我们已经知道了协程的启动、挂起、取消、异常以及常用的协程作用域等基础应用。 这些基础应用适合的场景是一次性任务执行完就结束了的场景。
launch / async 适合的场景
网络请求数据库查询文件读写并行计算任务等等
Channel 适合的场景
而对于一些相对复杂的场景例如持续的数据流、需要在不同的协程之间传递数据、需要顺序或背压控制等场景基础的 launch / async 就不够用了。
例如
用户点击、输入等事件流的处理生产者-消费者模型的需求任务排队、日志流高频数据源处理相机帧、音频流等
类似这种持续的、需要顺序控制、或者多个协程配合执行的场景就需要用到 Channel 了。 Channel 的概念和基本使用
概念
顾名思义Channel 有管道、通道的意思。Channel 跟 Java 中的 BlockingQueue 很相似区别在于 Channel 是挂起的不是阻塞的。
Channel 的核心特点就是能够在不同的协程之间进行数据传递并且能够控制数据传递的顺序。 使用起来很简单基本就分为以下几步
创建 Channel通过 channel.send 发送数据通过 channel.receive 接收数据
整体的概念也比较简单形象就是一根管道一个口子发送数据一个口子接收数据。
Channel 的创建
先来看下 Channel 的源码可以看到会根据传入的参数选择不同的实现。
public fun E Channel(capacity: Int RENDEZVOUS,onBufferOverflow: BufferOverflow BufferOverflow.SUSPEND,onUndeliveredElement: ((E) - Unit)? null
): ChannelE when (capacity) {RENDEZVOUS - {if (onBufferOverflow BufferOverflow.SUSPEND)BufferedChannel(RENDEZVOUS, onUndeliveredElement) // an efficient implementation of rendezvous channelelseConflatedBufferedChannel(1,onBufferOverflow,onUndeliveredElement) // support buffer overflow with buffered channel}CONFLATED - {require(onBufferOverflow BufferOverflow.SUSPEND) {CONFLATED capacity cannot be used with non-default onBufferOverflow}ConflatedBufferedChannel(1, BufferOverflow.DROP_OLDEST, onUndeliveredElement)}UNLIMITED - BufferedChannel(UNLIMITED,onUndeliveredElement) // ignores onBufferOverflow: it has buffer, but it never overflowsBUFFERED - { // uses default capacity with SUSPENDif (onBufferOverflow BufferOverflow.SUSPEND) BufferedChannel(CHANNEL_DEFAULT_CAPACITY,onUndeliveredElement)else ConflatedBufferedChannel(1, onBufferOverflow, onUndeliveredElement)}else - {if (onBufferOverflow BufferOverflow.SUSPEND) BufferedChannel(capacity, onUndeliveredElement)else ConflatedBufferedChannel(capacity, onBufferOverflow, onUndeliveredElement)}}
参数概览
参数类型默认值描述capacityIntRENDEZVOUS通道容量决定缓冲区大小和行为模式onBufferOverflowBufferOverflowSUSPEND缓冲区溢出时的处理策略onUndeliveredElement((E) - Unit)?null元素未能送达时的回调函数
capacity容量配置
capacity 参数决定了 Channel 的缓冲行为和容量大小
RENDEZVOUS值为 0无缓冲发送者和接收者必须同时准备好CONFLATED值为 -1只保留最新的元素旧元素会被覆盖UNLIMITED值为 Int.MAX_VALUE理论上就是无限容量永不阻塞发送BUFFERED值为 64默认缓冲大小自定义正整数自己指定具体的缓冲区大小
onBufferOverflow溢出策略
当缓冲区满时的处理策略
SUSPEND挂起发送操作等待缓冲区有空间默认DROP_OLDEST丢弃旧的元素添加新元素DROP_LATEST丢弃新元素保留缓冲区中的现有元素
onUndeliveredElement未送达回调
当元素无法送达时的清理回调函数
null不执行任何清理操作默认自定义函数用于资源清理、日志记录等根据业务需求来定义
参数组合效果
capacityonBufferOverflow行为适用场景RENDEZVOUSSUSPEND无缓冲同步通信严格的生产者-消费者同步BUFFEREDSUSPEND有限缓冲满时挂起一般的异步处理默认的缓冲数量是 64UNLIMITEDSUSPEND缓冲长度为 Int.MAX_VALUE高吞吐量场景生产上不建议使用有内存方面的风险CONFLATEDDROP_OLDEST无缓冲只保留最新值状态更新、实时数据自定义大小SUSPEND固定大小满时挂起批量处理、批量任务自定义大小DROP_OLDEST固定大小丢弃旧数据获取最近 N 个元素自定义大小DROP_LATEST固定大小拒绝新数据保护重要历史数据Capacity
RENDEZVOUS会合模式
特点
容量为 0无缓冲区发送者和接收者必须同时准备好才能完成数据传输提供强同步保证一手交钱一手交货
使用示例
suspend fun demonstrateRendezvousChannel() {// 创建 RENDEZVOUS Channel默认容量为 0默认什么都不传就是 rendezvous 模式ChannelString()val rendezvousChannel ChannelString(Channel.RENDEZVOUS)// 启动发送者协程val senderJob GlobalScope.launch {println([发送者] 准备发送消息...)rendezvousChannel.send(Hello from RENDEZVOUS!)println([发送者] 消息已发送)rendezvousChannel.send(Second message)println([发送者] 第二条消息已发送)rendezvousChannel.close()}// 启动接收者协程val receiverJob GlobalScope.launch {delay(1000) // 延迟1秒发送者会等待接收者准备好println([接收者] 开始接收消息...)for (message in rendezvousChannel) {println([接收者] 收到消息: $message)delay(500) // 模拟处理时间}println([接收者] Channel已关闭)}// 等待所有协程完成joinAll(senderJob, receiverJob)
}
执行结果
CONFLATED只留最新值
特点
容量为 1但会丢弃旧值只保留最新的元素发送操作永不阻塞只能使用 BufferOverflow.SUSPEND 策略
源码分析
CONFLATED - {require(onBufferOverflow BufferOverflow.SUSPEND) {CONFLATED capacity cannot be used with non-default onBufferOverflow}ConflatedBufferedChannel(1, BufferOverflow.DROP_OLDEST, onUndeliveredElement)
}使用示例
suspend fun demonstrateConflatedChannel() {// 创建 CONFLATED Channel相当于ChannelString(1, BufferOverflow.DROP_OLDEST)val conflatedChannel ChannelString(Channel.CONFLATED)// 快速发送多个消息val senderJob GlobalScope.launch {repeat(5) { i -val message Update-$iconflatedChannel.send(message)println([发送者] 发送更新: $message)delay(100) // 短暂延迟}conflatedChannel.close()}// 慢速接收者val receiverJob GlobalScope.launch {delay(1000) // 延迟1秒让发送者发送完所有消息println([接收者] 开始接收只会收到最新的值...)for (message in conflatedChannel) {println([接收者] 收到: $message)}}joinAll(senderJob, receiverJob)
} UNLIMITED无限容量
特点
容量为 Int.MAX_VALUE理论上无限容量发送操作永不阻塞但要注意内存使用忽略 onBufferOverflow 参数适用于高吞吐量场景但生产环境需谨慎使用
suspend fun demonstrateUnlimitedChannel() {val unlimitedChannel ChannelString(Channel.UNLIMITED)val senderJob GlobalScope.launch {repeat(10) { i -val message Message-$iunlimitedChannel.send(message)println([发送者] 立即发送: $message)}unlimitedChannel.close()println([发送者] 所有消息已发送Channel已关闭)}val receiverJob GlobalScope.launch {delay(1000) // 延迟1秒开始接收println([接收者] 开始慢速接收...)for (message in unlimitedChannel) {println([接收者] 处理: $message)delay(300) // 模拟处理时间}}joinAll(senderJob, receiverJob)
} BUFFERED有限容量
特点
使用默认容量 (CHANNEL_DEFAULT_CAPACITY通常为 64)在缓冲区满时根据 onBufferOverflow 策略处理
源码分析
BUFFERED - {if (onBufferOverflow BufferOverflow.SUSPEND)BufferedChannel(CHANNEL_DEFAULT_CAPACITY, onUndeliveredElement)elseConflatedBufferedChannel(1, onBufferOverflow, onUndeliveredElement)
}使用示例
suspend fun demonstrateBufferedDefaultChannel() {// 创建 BUFFERED Channel默认容量为 64val bufferedChannel ChannelString(Channel.BUFFERED)val senderJob GlobalScope.launch {repeat(100) { i -bufferedChannel.send(Message-$i)println([发送者] 发送 Message-$i)}bufferedChannel.close()}val receiverJob GlobalScope.launch {delay(1000) // 延迟接收for (message in bufferedChannel) {println([接收者] 收到: $message)delay(50)}}joinAll(senderJob, receiverJob)
}与下面自定义容量效果类似。
自定义容量
特点
指定具体的缓冲区大小根据 onBufferOverflow 策略处理溢出
源码分析
else - {if (onBufferOverflow BufferOverflow.SUSPEND)BufferedChannel(capacity, onUndeliveredElement)elseConflatedBufferedChannel(capacity, onBufferOverflow, onUndeliveredElement)
}使用示例
suspend fun demonstrateBufferedChannel() {// 创建容量为3的缓冲Channelval bufferedChannel ChannelInt(capacity 3)// 启动发送者协程val senderJob GlobalScope.launch {repeat(5) { i -println([发送者] 发送数字: $i)bufferedChannel.send(i)println([发送者] 数字 $i 已发送)}bufferedChannel.close()println([发送者] Channel已关闭)}// 启动接收者协程延迟接收以观察缓冲效果val receiverJob GlobalScope.launch {delay(2000) // 延迟2秒开始接收println([接收者] 开始接收数字...)for (number in bufferedChannel) {println([接收者] 收到数字: $number)delay(800) // 模拟慢速处理}}joinAll(senderJob, receiverJob)
}可以看到因为默认的溢出策略是 SUSPEND所以当缓冲区满了时发送者会被挂起直到接收者处理完一个元素才会继续发送。 BufferOverflow 策略详解
当 Channel 的缓冲区满时BufferOverflow 参数决定了如何处理新的发送请求
SUSPEND默认策略
行为当缓冲区满时挂起发送操作直到有空间可用特点提供背压控制防止生产者过快使用场景需要确保所有数据都被处理的场景
suspend fun demonstrateBasicOperations() {//容量为 2溢出策略为SUSPENDval channel ChannelString(capacity 2, onBufferOverflow BufferOverflow.SUSPEND)//发送的速度快val job1 GlobalScope.launch {repeat(5) {channel.send(Message-$it)println([发送者] 发送 Message-$it)}channel.close()}val job2 GlobalScope.launch {//除了用 channel.recrive 外也可以直接 用 for 循环接收数据for (message in channel) {//接收的速度慢delay(1000)println([接收者] 接收到: $message)}}joinAll(job1, job2)
} DROP_LATEST
行为当缓冲区满时丢弃新元素保留缓冲区中的现有元素特点保护已缓冲的数据不被覆盖使用场景保护重要的历史数据防止新数据覆盖性能特点发送操作永不阻塞但新数据可能被丢弃
suspend fun demonstrateBasicOperations() {val channel ChannelString(capacity 2, onBufferOverflow BufferOverflow.DROP_LATEST)val job1 GlobalScope.launch {repeat(5) {channel.send(Message-$it)println([发送者] 发送 Message-$it)}channel.close()}val job2 GlobalScope.launch {for (message in channel) {delay(1000)println([接收者] 接收到: $message)}}joinAll(job1, job2)
}可以看到当缓冲区满时会把新数据丢弃掉因此接收端只接收到了旧数据。 DROP_OLDEST
行为当缓冲区满时丢弃旧的元素添加新元素特点保持固定的内存使用优先保留新数据使用场景实时数据流、最近N个元素性能特点发送操作永不阻塞但可能丢失历史数据
suspend fun demonstrateBasicOperations() {val channel ChannelString(capacity 2, onBufferOverflow BufferOverflow.DROP_OLDEST)val job1 GlobalScope.launch {repeat(5) {channel.send(Message-$it)println([发送者] 发送 Message-$it)}channel.close()}val job2 GlobalScope.launch {for (message in channel) {delay(1000)println([接收者] 接收到: $message)}}joinAll(job1, job2)
}需要注意的是当缓冲区满了之后1 和 2 被丢弃了3 和 4 被放进去了。从这里可以看出丢弃数据时并不是把最早的旧数据丢掉这里跟内部的实现有关。 onUndeliveredElement 回调
当元素无法送达时如 Channel 被取消或关闭会调用此回调函数
suspend fun demonstrateBasicOperations() {val channel ChannelString(capacity 2, onBufferOverflow BufferOverflow.DROP_OLDEST) {println([Channel] 缓冲区已满无法放到缓冲区,值${it})}// 演示基本的send和receive操作val job1 GlobalScope.launch {repeat(5) {channel.send(Message-$it)println([发送者] 发送 Message-$it)}channel.close()}val job2 GlobalScope.launch {for (message in channel) {delay(1000)println([接收者] 接收到: $message)}}joinAll(job1, job2)
} Channel 操作方式
Channel 提供了两种操作方式阻塞操作和非阻塞操作。
阻塞操作send/receive
send() 和 receive() 方法都是挂起方法它们会阻塞当前协程直到完成操作。
非阻塞操作trySend/tryReceive
trySend() 和 tryReceive() 是 Channel 提供的非阻塞操作 API。与阻塞版本不同这些方法会立即返回结果不会挂起当前协程也不会抛出异常。
操作对比
操作类型阻塞版本非阻塞版本行为差异发送send()trySend()send() 会挂起直到有空间trySend() 立即返回结果接收receive()tryReceive()receive() 会挂起直到有数据tryReceive() 立即返回结果
返回值类型
trySend() 返回 ChannelResultUnittryReceive() 返回 ChannelResultT
ChannelResult 是一个密封类通过密封类中的成员 isSuccess 和 getOrNull() 可以判断操作是否成功。 大部分场景下send / receive 合理的 Channel 配置就能解决问题trySend/tryReceive 更多的是想达到如下效果
避免不必要的协程挂起开销希望立即得到结果提供更精细的控制逻辑如超时处理、重试机制等实现更好的错误处理和用户反馈能更好地处理异常场景 runBlocking {val channel ChannelInt(2)val sendJob launch {repeat(5) {delay(100)val sendResult channel.trySend(it)sendResult.onSuccess {println(发送成功)}.onFailure {println(发送失败)}.onClosed {println(通道已关闭)}}}val receiveJob launch {for (i in channel) {delay(300)println(接收到数据:${i})}}joinAll(sendJob, receiveJob)}Channel 状态管理
Channel 在其生命周期中会经历以下几个关键状态
活跃状态Active可以正常发送和接收数据发送端关闭Closed for Send不能发送新数据但可以接收缓冲区中的数据接收端关闭Closed for Receive不能接收数据缓冲区已清空取消状态CancelledChannel 被取消所有操作都会失败
API
channel.close()关闭 Channelchannel.isClosedForSend判断发送端是否已关闭channel.isClosedForReceive判断接收端是否已关闭channel.cancel()取消 Channel
Close关闭操作
调用 close() 后isClosedForSend 立即变为 true此时缓冲区中的数据仍可被消费只有当缓冲区清空后isClosedForReceive 才变为 true
示例 suspend fun demonstrateChannelClose() {val channel ChannelString(1)val producer GlobalScope.launch {try {for (i in 1..5) {val message Message $iprintln(准备发送: $message)channel.send(message)println(成功发送: $message)delay(100)}} catch (e: ClosedSendChannelException) {println(生产者: Channel已关闭无法发送数据 - ${e.message})}}val consumer GlobalScope.launch {try {for (message in channel) {println(接收到: $message)delay(200)}println(消费者: Channel已关闭退出接收循环)} catch (e: Exception) {println(消费者异常: ${e.message})}}delay(300) // 模拟让一些数据能够被接收到// 检查Channel状态println(关闭前状态:)println( isClosedForSend: ${channel.isClosedForSend})println( isClosedForReceive: ${channel.isClosedForReceive})// 关闭Channelprintln(\n正在关闭Channel...)channel.close()// 检查关闭后的状态println(关闭后状态:)println( isClosedForSend: ${channel.isClosedForSend})println( isClosedForReceive: ${channel.isClosedForReceive})// 等待协程完成producer.join()consumer.join()println(最终状态:)println( isClosedForSend: ${channel.isClosedForSend})println( isClosedForReceive: ${channel.isClosedForReceive})
} Cancel取消操作
cancel() 方法用于强制取消 Channel它会
立即关闭发送和接收端清空缓冲区中的所有数据触发 onUndeliveredElement 回调如果设置了
suspend fun demonstrateChannelCancel() {val channel ChannelString(capacity 5) {println(消息未被接收:${it})}val producer GlobalScope.launch {try {for (i in 1..8) {val message Message $iprintln(尝试发送: $message)channel.send(message)println(成功发送: $message)delay(100)}} catch (e: CancellationException) {println(生产者: Channel被取消 - ${e.message})}}val consumer GlobalScope.launch {try {for (message in channel) {println(接收到: $message)delay(300)}} catch (e: CancellationException) {println(消费者: 协程被取消 - ${e.message})}}delay(400) // 让一些操作执行println(\n取消前状态:)println( isClosedForSend: ${channel.isClosedForSend})println( isClosedForReceive: ${channel.isClosedForReceive})// 取消Channelprintln(\n正在取消Channel...)channel.cancel(CancellationException(主动取消Channel))println(取消后状态:)println( isClosedForSend: ${channel.isClosedForSend})println( isClosedForReceive: ${channel.isClosedForReceive})// 等待协程完成producer.join()consumer.join()
} Channel 异常处理
在使用 Channel 的过程中会遇到各种异常情况。主要包括以下几种类型
ClosedSendChannelException
触发条件
在已关闭的 Channel 上调用 send() 方法Channel 调用 close() 后发送端立即关闭
示例
suspend fun demonstrateClosedSendException() {val channel ChannelString()// 关闭 Channelchannel.close()try {// 尝试在已关闭的 Channel 上发送数据channel.send(This will throw exception)} catch (e: ClosedSendChannelException) {println(捕获异常: ${e.message})println(异常类型: ${e::class.simpleName})}
}ClosedReceiveChannelException
触发条件
从已关闭且缓冲区为空的 Channel 调用 receive() 方法当 isClosedForReceive 为 true 时调用 receive()
示例
suspend fun demonstrateClosedReceiveException() {val channel ChannelString()// 关闭 Channelchannel.close()try {// 尝试从已关闭且空的 Channel 接收数据val message channel.receive()println(收到消息: $message)} catch (e: ClosedReceiveChannelException) {println(捕获异常: ${e.message})println(异常类型: ${e::class.simpleName})}
}CancellationException
触发条件
Channel 被 cancel() 方法取消父协程被取消导致 Channel 操作被取消超时或其他取消信号
示例
suspend fun demonstrateCancellationException() {val channel ChannelString()val job GlobalScope.launch {try {// 这个操作会被取消channel.send(This will be cancelled)} catch (e: CancellationException) {println(发送操作被取消: ${e.message})throw e // 重新抛出 CancellationException}}delay(100)// 取消 Channelchannel.cancel(CancellationException(手动取消 Channel))try {job.join()} catch (e: CancellationException) {println(协程被取消: ${e.message})}
}异常与状态关系
Channel 状态send() 行为receive() 行为trySend() 行为tryReceive() 行为活跃状态正常发送或挂起正常接收或挂起返回成功/失败结果返回成功/失败结果发送端关闭抛出 ClosedSendChannelException正常接收缓冲区数据返回失败结果正常返回结果接收端关闭抛出 ClosedSendChannelException抛出 ClosedReceiveChannelException返回失败结果返回失败结果已取消抛出 CancellationException抛出 CancellationException返回失败结果返回失败结果
异常处理技巧
使用非阻塞操作避免异常
非阻塞操作不会抛出异常而是返回结果对象
suspend fun safeChannelOperations() {val channel ChannelString()// 安全的发送操作val sendResult channel.trySend(Safe message)when {sendResult.isSuccess - println(发送成功)sendResult.isFailure - println(发送失败: ${sendResult.exceptionOrNull()})sendResult.isClosed - println(Channel 已关闭)}// 安全的接收操作val receiveResult channel.tryReceive()when {receiveResult.isSuccess - println(接收到: ${receiveResult.getOrNull()})receiveResult.isFailure - println(接收失败: ${receiveResult.exceptionOrNull()})receiveResult.isClosed - println(Channel 已关闭)}
}健壮的异常处理
suspend fun robustChannelUsage() {val channel ChannelString()val producer GlobalScope.launch {try {repeat(5) { i -if (channel.isClosedForSend) {println(Channel 已关闭停止发送)break}channel.send(Message $i)delay(100)}} catch (e: ClosedSendChannelException) {println(生产者: Channel 已关闭)} catch (e: CancellationException) {println(生产者: 操作被取消)throw e // 重新抛出取消异常} finally {println(生产者: 清理资源)}}val consumer GlobalScope.launch {try {while (!channel.isClosedForReceive) {try {val message channel.receive()println(消费者: 收到 $message)} catch (e: ClosedReceiveChannelException) {println(消费者: Channel 已关闭且无更多数据)break}delay(200)}} catch (e: CancellationException) {println(消费者: 操作被取消)throw e} finally {println(消费者: 清理资源)}}delay(1000)channel.close()joinAll(producer, consumer)
}总结
Channel 关键概念对比
特性RENDEZVOUSCONFLATEDBUFFEREDUNLIMITED自定义容量容量0164Int.MAX_VALUE指定值缓冲行为无缓冲同步只保留最新值有限缓冲无限缓冲有限缓冲发送阻塞是否缓冲满时否缓冲满时适用场景严格同步状态更新一般异步高吞吐量批量处理内存风险低低中等高可控
溢出策略对比
策略行为性能特点适用场景SUSPEND挂起发送操作提供背压控制确保数据完整性DROP_OLDEST丢弃旧元素发送不阻塞实时数据流DROP_LATEST丢弃新元素发送不阻塞保护历史数据
操作方式
操作类型阻塞版本非阻塞版本异常处理返回值发送send()trySend()抛出异常ChannelResultUnit接收receive()tryReceive()抛出异常ChannelResultT特点会挂起协程立即返回需要 try-catch通过结果对象判断
Channel 状态生命周期
状态描述send()receive()检查方法活跃正常工作状态✅ 正常✅ 正常-发送关闭调用 close() 后❌ 异常✅ 可接收缓冲区数据isClosedForSend接收关闭缓冲区清空后❌ 异常❌ 异常isClosedForReceive已取消调用 cancel() 后❌ 异常❌ 异常-
总体来说Channel 是一种非常强大的协程通信机制它可以帮助我们在协程之间进行安全、高效的通信。在使用 Channel时我们需要注意异常处理、缓冲区容量、溢出策略等问题。 感谢阅读如果对你有帮助请三连点赞、收藏、加关注支持。有任何疑问或建议欢迎在评论区留言讨论。如需转载请注明出处喻志强的博客
