苏州银行 發表於 2025-8-22 09:34:17

Kotlin 协程之Channel的概念和基本使用详解

<div id="navCategory"><h5 class="catalogue">目录</h5><ul class="first_class_ul"><li>前言</li><ul class="second_class_ul"><li>launch / async 适合的场景</li></ul><li>Channel 的概念和基本使用</li><ul class="second_class_ul"><li>概念</li><li>Channel 的创建</li><li>参数概览</li><ul class="third_class_ul"><li>capacity（容量配置）</li><li>onBufferOverflow（溢出策略）</li><li>onUndeliveredElement（未送达回调）</li></ul><li>参数组合效果</li><ul class="third_class_ul"></ul></ul><li>Capacity</li><ul class="second_class_ul"><li>RENDEZVOUS（会合模式）</li><ul class="third_class_ul"></ul><li>CONFLATED（只留最新值）</li><ul class="third_class_ul"></ul><li>UNLIMITED（无限容量）</li><ul class="third_class_ul"></ul><li>BUFFERED（有限容量）</li><ul class="third_class_ul"></ul><li>自定义容量</li><ul class="third_class_ul"></ul></ul><li>BufferOverflow 策略详解</li><ul class="second_class_ul"><li>SUSPEND（默认策略）</li><ul class="third_class_ul"></ul><li>DROP_LATEST</li><ul class="third_class_ul"></ul><li>DROP_OLDEST</li><ul class="third_class_ul"></ul></ul><li>onUndeliveredElement 回调</li><ul class="second_class_ul"></ul><li>Channel 操作方式</li><ul class="second_class_ul"><li>阻塞操作（send/receive）</li><ul class="third_class_ul"></ul><li>非阻塞操作（trySend/tryReceive）</li><ul class="third_class_ul"></ul><li>操作对比</li><ul class="third_class_ul"></ul><li>返回值类型</li><ul class="third_class_ul"></ul></ul><li>Channel 状态管理</li><ul class="second_class_ul"><li>API</li><ul class="third_class_ul"></ul><li>Close（关闭操作）</li><ul class="third_class_ul"></ul><li>Cancel（取消操作）</li><ul class="third_class_ul"></ul></ul><li>Channel 异常处理</li><ul class="second_class_ul"><li>ClosedSendChannelException</li><ul class="third_class_ul"></ul><li>ClosedReceiveChannelException</li><ul class="third_class_ul"></ul><li>CancellationException</li><ul class="third_class_ul"></ul><li>异常与状态关系</li><ul class="third_class_ul"></ul><li>异常处理技巧</li><ul class="third_class_ul"><li>使用非阻塞操作避免异常</li><li>健壮的异常处理</li></ul></ul><li>总结</li><ul class="second_class_ul"><li>Channel 关键概念对比</li><ul class="third_class_ul"></ul><li>溢出策略对比</li><ul class="third_class_ul"></ul><li>操作方式</li><ul class="third_class_ul"></ul><li>Channel 状态生命周期</li><ul class="third_class_ul"></ul></ul></ul></div><p class="maodian"></p><h2>前言</h2>
<p>在 专栏 之前的文章中，我们已经知道了协程的启动、挂起、取消、异常以及常用的协程作用域等基础应用。<br />这些基础应用适合的场景是一次性任务，执行完就结束了的场景。</p>
<p class="maodian"></p><h3>launch / async 适合的场景</h3>
<ul><li>网络请求</li><li>数据库查询</li><li>文件读写</li><li>并行计算任务</li><li>等等</li></ul>
<p>而对于一些相对复杂的场景，例如：持续的数据流、需要在不同的协程之间传递数据、需要顺序或背压控制等场景，基础的 launch / async<br />就不够用了。</p>
<p>例如：</p>
<ul><li>用户点击、输入等事件流的处理</li><li>生产者-消费者模型的需求：任务排队、日志流</li><li>高频数据源处理（相机帧、音频流等）</li></ul>
<p>类似这种持续的、需要顺序控制、或者多个协程配合执行的场景，就需要用到 <code>Channel</code> 了。</p>
<p class="maodian"></p><h2>Channel 的概念和基本使用</h2>
<p class="maodian"></p><h3>概念</h3>
<p>顾名思义，<code>Channel</code> 有管道、通道的意思。<code>Channel</code> 跟 Java 中的 <code>BlockingQueue</code> 很相似，区别在于 <code>Channel</code> 是挂起的，不是阻塞的。</p>
<p><code>Channel</code> 的核心特点就是能够在不同的协程之间进行数据传递，并且能够控制数据传递的顺序。<br />使用起来很简单，基本就分为以下几步：</p>
<ol><li>创建 Channel</li><li>通过 <code>channel.send</code> 发送数据</li><li>通过 <code>channel.receive</code> 接收数据</li></ol>
<p>整体的概念也比较简单形象，就是一根管道，一个口子发送数据，一个口子接收数据。</p>
<p class="maodian"></p><h3>Channel 的创建</h3>
<p>先来看下 Channel 的源码，可以看到会根据传入的参数选择不同的实现。</p>
<div class="jb51code"><pre class="brush:java;">public fun &lt;E&gt; Channel(
    capacity: Int = RENDEZVOUS,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    onUndeliveredElement: ((E) -&gt; Unit)? = null
): Channel&lt;E&gt; =
    when (capacity) {
      RENDEZVOUS -&gt; {
            if (onBufferOverflow == BufferOverflow.SUSPEND)
                BufferedChannel(RENDEZVOUS, onUndeliveredElement) // an efficient implementation of rendezvous channel
            else
                ConflatedBufferedChannel(
                  1,
                  onBufferOverflow,
                  onUndeliveredElement
                ) // support buffer overflow with buffered channel
      }
      CONFLATED -&gt; {
            require(onBufferOverflow == BufferOverflow.SUSPEND) {
                "CONFLATED capacity cannot be used with non-default onBufferOverflow"
            }
            ConflatedBufferedChannel(1, BufferOverflow.DROP_OLDEST, onUndeliveredElement)
      }
      UNLIMITED -&gt; BufferedChannel(
            UNLIMITED,
            onUndeliveredElement
      ) // ignores onBufferOverflow: it has buffer, but it never overflows
      BUFFERED -&gt; { // uses default capacity with SUSPEND
            if (onBufferOverflow == BufferOverflow.SUSPEND) BufferedChannel(
                CHANNEL_DEFAULT_CAPACITY,
                onUndeliveredElement
            )
            else ConflatedBufferedChannel(1, onBufferOverflow, onUndeliveredElement)
      }
      else -&gt; {
            if (onBufferOverflow === BufferOverflow.SUSPEND) BufferedChannel(capacity, onUndeliveredElement)
            else ConflatedBufferedChannel(capacity, onBufferOverflow, onUndeliveredElement)
      }
    }</pre></div>
<p class="maodian"></p><h3>参数概览</h3>
<table><thead><tr><th>参数</th><th>类型</th><th>默认值</th><th>描述</th></tr></thead><tbody><tr><td><code>capacity</code></td><td><code>Int</code></td><td><code>RENDEZVOUS</code></td><td>通道容量，决定缓冲区大小和行为模式</td></tr><tr><td><code>onBufferOverflow</code></td><td><code>BufferOverflow</code></td><td><code>SUSPEND</code></td><td>缓冲区溢出时的处理策略</td></tr><tr><td><code>onUndeliveredElement</code></td><td><code>((E) -&gt; Unit)?</code></td><td><code>null</code></td><td>元素未能送达时的回调函数</td></tr></tbody></table>
<p class="maodian"></p><h4>capacity（容量配置）</h4>
<p><code>capacity</code> 参数决定了 <code>Channel</code> 的缓冲行为和容量大小：</p>
<ul><li><code>RENDEZVOUS</code>（值为 0）：无缓冲，发送者和接收者必须同时准备好</li><li><code>CONFLATED</code>（值为 -1）：只保留最新的元素，旧元素会被覆盖</li><li><code>UNLIMITED</code>（值为 <code>Int.MAX_VALUE</code>）：理论上就是无限容量，永不阻塞发送</li><li><code>BUFFERED</code>（值为 64）：默认缓冲大小</li><li>自定义正整数：自己指定具体的缓冲区大小</li></ul>
<p class="maodian"></p><h4>onBufferOverflow（溢出策略）</h4>
<p>当缓冲区满时的处理策略：</p>
<ul><li><code>SUSPEND</code>：挂起发送操作，等待缓冲区有空间（默认）</li><li><code>DROP_OLDEST</code>：丢弃旧的元素，添加新元素</li><li><code>DROP_LATEST</code>：丢弃新元素，保留缓冲区中的现有元素</li></ul>
<p class="maodian"></p><h4>onUndeliveredElement（未送达回调）</h4>
<p>当元素无法送达时的清理回调函数：</p>
<ul><li><code>null</code>：不执行任何清理操作（默认）</li><li><strong>自定义函数</strong>：用于资源清理、日志记录等，根据业务需求来定义</li></ul>
<p class="maodian"></p><h3>参数组合效果</h3>
<table><thead><tr><th>capacity</th><th>onBufferOverflow</th><th>行为</th><th>适用场景</th></tr></thead><tbody><tr><td><code>RENDEZVOUS</code></td><td><code>SUSPEND</code></td><td>无缓冲，同步通信</td><td>严格的生产者-消费者同步</td></tr><tr><td><code>BUFFERED</code></td><td><code>SUSPEND</code></td><td>有限缓冲，满时挂起</td><td>一般的异步处理，默认的缓冲数量是 64</td></tr><tr><td><code>UNLIMITED</code></td><td><code>SUSPEND</code></td><td>缓冲长度为 Int.MAX_VALUE</td><td>高吞吐量场景（生产上不建议使用，有内存方面的风险）</td></tr><tr><td><code>CONFLATED</code></td><td><code>DROP_OLDEST</code></td><td>无缓冲，只保留最新值</td><td>状态更新、实时数据</td></tr><tr><td>自定义大小</td><td><code>SUSPEND</code></td><td>固定大小，满时挂起</td><td>批量处理、批量任务</td></tr><tr><td>自定义大小</td><td><code>DROP_OLDEST</code></td><td>固定大小，丢弃旧数据</td><td>获取最近 N 个元素</td></tr><tr><td>自定义大小</td><td><code>DROP_LATEST</code></td><td>固定大小，拒绝新数据</td><td>保护重要历史数据</td></tr></tbody></table>
<p class="maodian"></p><h2>Capacity</h2>
<p class="maodian"></p><h3>RENDEZVOUS（会合模式）</h3>
<p><strong>特点：</strong></p>
<ul><li>容量为 0，无缓冲区</li><li>发送者和接收者必须同时准备好才能完成数据传输</li><li>提供强同步保证，一手交钱一手交货</li></ul>
<p><strong>使用示例：</strong></p>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateRendezvousChannel() {
    // 创建 RENDEZVOUS Channel（默认容量为 0），默认什么都不传就是 rendezvous 模式，Channel&lt;String&gt;()
    val rendezvousChannel = Channel&lt;String&gt;(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)
}</pre></div>
<p>执行结果</p>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244620.png" /></p>
<p class="maodian"></p><h3>CONFLATED（只留最新值）</h3>
<p><strong>特点：</strong></p>
<ul><li>容量为 1，但会丢弃旧值</li><li>只保留最新的元素</li><li>发送操作永不阻塞</li><li>只能使用 <code>BufferOverflow.SUSPEND</code> 策略</li></ul>
<p><strong>源码分析：</strong></p>
<div class="jb51code"><pre class="brush:java;">CONFLATED -&gt; {
    require(onBufferOverflow == BufferOverflow.SUSPEND) {
      "CONFLATED capacity cannot be used with non-default onBufferOverflow"
    }
    ConflatedBufferedChannel(1, BufferOverflow.DROP_OLDEST, onUndeliveredElement)
}</pre></div>
<p><strong>使用示例：</strong></p>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateConflatedChannel() {
    // 创建 CONFLATED Channel，相当于：Channel&lt;String&gt;(1, BufferOverflow.DROP_OLDEST)
    val conflatedChannel = Channel&lt;String&gt;(Channel.CONFLATED)
    // 快速发送多个消息
    val senderJob = GlobalScope.launch {
      repeat(5) { i -&gt;
            val message = "Update-$i"
            conflatedChannel.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)
}</pre></div>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244698.png" /></p>
<p class="maodian"></p><h3>UNLIMITED（无限容量）</h3>
<p><strong>特点：</strong></p>
<ul><li>容量为 <code>Int.MAX_VALUE</code>，理论上无限容量</li><li>发送操作永不阻塞，但要注意内存使用</li><li>忽略 <code>onBufferOverflow</code> 参数</li><li>适用于高吞吐量场景，但生产环境需谨慎使用</li></ul>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateUnlimitedChannel() {
    val unlimitedChannel = Channel&lt;String&gt;(Channel.UNLIMITED)
    val senderJob = GlobalScope.launch {
      repeat(10) { i -&gt;
            val message = "Message-$i"
            unlimitedChannel.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)
}</pre></div>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244697.jpg" /></p>
<p class="maodian"></p><h3>BUFFERED（有限容量）</h3>
<p><strong>特点：</strong></p>
<ul><li>使用默认容量 (<code>CHANNEL_DEFAULT_CAPACITY</code>，通常为 64)</li><li>在缓冲区满时根据 <code>onBufferOverflow</code> 策略处理</li></ul>
<p><strong>源码分析：</strong></p>
<div class="jb51code"><pre class="brush:java;">BUFFERED -&gt; {
    if (onBufferOverflow == BufferOverflow.SUSPEND)
      BufferedChannel(CHANNEL_DEFAULT_CAPACITY, onUndeliveredElement)
    else
      ConflatedBufferedChannel(1, onBufferOverflow, onUndeliveredElement)
}</pre></div>
<p><strong>使用示例：</strong></p>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateBufferedDefaultChannel() {
    // 创建 BUFFERED Channel（默认容量为 64）
    val bufferedChannel = Channel&lt;String&gt;(Channel.BUFFERED)
    val senderJob = GlobalScope.launch {
      repeat(100) { i -&gt;
            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)
}</pre></div>
<p>与下面自定义容量效果类似。</p>
<p class="maodian"></p><h3>自定义容量</h3>
<p><strong>特点：</strong></p>
<ul><li>指定具体的缓冲区大小</li><li>根据 <code>onBufferOverflow</code> 策略处理溢出</li></ul>
<p><strong>源码分析：</strong></p>
<div class="jb51code"><pre class="brush:java;">else -&gt; {
    if (onBufferOverflow === BufferOverflow.SUSPEND)
      BufferedChannel(capacity, onUndeliveredElement)
    else
      ConflatedBufferedChannel(capacity, onBufferOverflow, onUndeliveredElement)
}</pre></div>
<p><strong>使用示例：</strong></p>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateBufferedChannel() {
    // 创建容量为3的缓冲Channel
    val bufferedChannel = Channel&lt;Int&gt;(capacity = 3)
    // 启动发送者协程
    val senderJob = GlobalScope.launch {
      repeat(5) { i -&gt;
            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)
}</pre></div>
<p>可以看到，因为默认的溢出策略是 <code>SUSPEND</code>，所以当缓冲区满了时，发送者会被挂起，直到接收者处理完一个元素，才会继续发送。</p>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244626.png" /></p>
<p class="maodian"></p><h2>BufferOverflow 策略详解</h2>
<p>当 Channel 的缓冲区满时，<code>BufferOverflow</code> 参数决定了如何处理新的发送请求：</p>
<p class="maodian"></p><h3>SUSPEND（默认策略）</h3>
<ul><li><strong>行为</strong>：当缓冲区满时，挂起发送操作直到有空间可用</li><li><strong>特点</strong>：提供背压控制，防止生产者过快</li><li><strong>使用场景</strong>：需要确保所有数据都被处理的场景</li></ul>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateBasicOperations() {
    //容量为 2，溢出策略为SUSPEND
    val channel = Channel&lt;String&gt;(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)
}</pre></div>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244667.png" /></p>
<p class="maodian"></p><h3>DROP_LATEST</h3>
<ul><li><strong>行为</strong>：当缓冲区满时，丢弃新元素，保留缓冲区中的现有元素</li><li><strong>特点</strong>：保护已缓冲的数据不被覆盖</li><li><strong>使用场景</strong>：保护重要的历史数据，防止新数据覆盖</li><li><strong>性能特点</strong>：发送操作永不阻塞，但新数据可能被丢弃</li></ul>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateBasicOperations() {
    val channel = Channel&lt;String&gt;(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)
}</pre></div>
<p>可以看到，当缓冲区满时，会把新数据丢弃掉，因此，接收端只接收到了旧数据。</p>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244647.png" /></p>
<p class="maodian"></p><h3>DROP_OLDEST</h3>
<ul><li><strong>行为</strong>：当缓冲区满时，丢弃旧的元素，添加新元素</li><li><strong>特点</strong>：保持固定的内存使用，优先保留新数据</li><li><strong>使用场景</strong>：实时数据流、最近N个元素</li><li><strong>性能特点</strong>：发送操作永不阻塞，但可能丢失历史数据</li></ul>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateBasicOperations() {
    val channel = Channel&lt;String&gt;(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)
}</pre></div>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244651.png" /></p>
<p>需要注意的是，当缓冲区满了之后，1 和 2 被丢弃了，3 和 4 被放进去了。从这里可以看出，丢弃数据时，并不是把最早的旧数据丢掉，这里跟内部的实现有关。</p>
<p class="maodian"></p><h2>onUndeliveredElement 回调</h2>
<p>当元素无法送达时（如 Channel 被取消或关闭），会调用此回调函数</p>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateBasicOperations() {
    val channel = Channel&lt;String&gt;(capacity = 2, onBufferOverflow = BufferOverflow.DROP_OLDEST) {
      println(" 缓冲区已满，无法放到缓冲区,值：${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)
}</pre></div>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244640.png" /></p>
<p class="maodian"></p><h2>Channel 操作方式</h2>
<p>Channel 提供了两种操作方式：阻塞操作和非阻塞操作。</p>
<p class="maodian"></p><h3>阻塞操作（send/receive）</h3>
<p><code>send()</code> 和 <code>receive()</code> 方法都是挂起方法，它们会阻塞当前协程，直到完成操作。</p>
<p class="maodian"></p><h3>非阻塞操作（trySend/tryReceive）</h3>
<p><code>trySend()</code> 和 <code>tryReceive()</code> 是 Channel 提供的非阻塞操作 API。与阻塞版本不同，这些方法会立即返回结果，不会挂起当前协程，也不会抛出异常。</p>
<p class="maodian"></p><h3>操作对比</h3>
<table><thead><tr><th>操作类型</th><th>阻塞版本</th><th>非阻塞版本</th><th>行为差异</th></tr></thead><tbody><tr><td>发送</td><td><code>send()</code></td><td><code>trySend()</code></td><td><code>send()</code> 会挂起直到有空间；<code>trySend()</code> 立即返回结果</td></tr><tr><td>接收</td><td><code>receive()</code></td><td><code>tryReceive()</code></td><td><code>receive()</code> 会挂起直到有数据；<code>tryReceive()</code> 立即返回结果</td></tr></tbody></table>
<p class="maodian"></p><h3>返回值类型</h3>
<ul><li><code>trySend()</code> 返回 <code>ChannelResult&lt;Unit&gt;</code></li><li><code>tryReceive()</code> 返回 <code>ChannelResult&lt;T&gt;</code></li></ul>
<p><code>ChannelResult</code> 是一个密封类，通过密封类中的成员 <code>isSuccess</code> 和 <code>getOrNull()</code> 可以判断操作是否成功。</p>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244634.jpg" /></p>
<p>大部分场景下，send / receive + 合理的 Channel 配置就能解决问题，trySend/tryReceive 更多的是想达到如下效果：</p>
<ul><li>避免不必要的协程挂起开销，希望立即得到结果</li><li>提供更精细的控制逻辑，如：超时处理、重试机制等</li><li>实现更好的错误处理和用户反馈，能更好地处理异常场景</li></ul>
<div class="jb51code"><pre class="brush:java;">runBlocking {
    val channel = Channel&lt;Int&gt;(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)
}</pre></div>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244667.png" /></p>
<p class="maodian"></p><h2>Channel 状态管理</h2>
<p>Channel 在其生命周期中会经历以下几个关键状态：</p>
<ul><li><strong>活跃状态（Active）</strong>：可以正常发送和接收数据</li><li><strong>发送端关闭（Closed for Send）</strong>：不能发送新数据，但可以接收缓冲区中的数据</li><li><strong>接收端关闭（Closed for Receive）</strong>：不能接收数据，缓冲区已清空</li><li><strong>取消状态（Cancelled）</strong>：Channel 被取消，所有操作都会失败</li></ul>
<p class="maodian"></p><h3>API</h3>
<ul><li><code>channel.close()</code>：关闭 Channel</li><li><code>channel.isClosedForSend</code>：判断发送端是否已关闭</li><li><code>channel.isClosedForReceive</code>：判断接收端是否已关闭</li><li><code>channel.cancel()</code>：取消 Channel</li></ul>
<p class="maodian"></p><h3>Close（关闭操作）</h3>
<ul><li>调用 <code>close()</code> 后，<code>isClosedForSend</code> 立即变为 <code>true</code></li><li>此时，缓冲区中的数据仍可被消费</li><li>只有当缓冲区清空后，<code>isClosedForReceive</code> 才变为 <code>true</code></li></ul>
<p>示例：</p>
<div class="jb51code"><pre class="brush:java;">    suspend fun demonstrateChannelClose() {
    val channel = Channel&lt;String&gt;(1)
    val producer = GlobalScope.launch {
      try {
            for (i in 1..5) {
                val message = "Message $i"
                println("准备发送: $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}")
    // 关闭Channel
    println("\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}")
}</pre></div>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244670.jpg" /></p>
<p class="maodian"></p><h3>Cancel（取消操作）</h3>
<p><code>cancel()</code> 方法用于强制取消 Channel，它会：</p>
<ul><li>立即关闭发送和接收端</li><li>清空缓冲区中的所有数据</li><li>触发 <code>onUndeliveredElement</code> 回调（如果设置了）</li></ul>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateChannelCancel() {
    val channel = Channel&lt;String&gt;(capacity = 5) {
      println("消息未被接收:${it}")
    }
    val producer = GlobalScope.launch {
      try {
            for (i in 1..8) {
                val message = "Message $i"
                println("尝试发送: $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}")
    // 取消Channel
    println("\n正在取消Channel...")
    channel.cancel(CancellationException("主动取消Channel"))
    println("取消后状态:")
    println("isClosedForSend: ${channel.isClosedForSend}")
    println("isClosedForReceive: ${channel.isClosedForReceive}")
    // 等待协程完成
    producer.join()
    consumer.join()
}</pre></div>
<p style="text-align:center"><img alt="" src="https://img.jbzj.com/file_images/article/202508/2025082209244673.jpg" /></p>
<p class="maodian"></p><h2>Channel 异常处理</h2>
<p>在使用 Channel 的过程中，会遇到各种异常情况。主要包括以下几种类型：</p>
<p class="maodian"></p><h3>ClosedSendChannelException</h3>
<p><strong>触发条件：</strong></p>
<ul><li>在已关闭的 Channel 上调用 <code>send()</code> 方法</li><li>Channel 调用 <code>close()</code> 后，发送端立即关闭</li></ul>
<p><strong>示例：</strong></p>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateClosedSendException() {
    val channel = Channel&lt;String&gt;()
    // 关闭 Channel
    channel.close()
    try {
      // 尝试在已关闭的 Channel 上发送数据
      channel.send("This will throw exception")
    } catch (e: ClosedSendChannelException) {
      println("捕获异常: ${e.message}")
      println("异常类型: ${e::class.simpleName}")
    }
}</pre></div>
<p class="maodian"></p><h3>ClosedReceiveChannelException</h3>
<p><strong>触发条件：</strong></p>
<ul><li>从已关闭且缓冲区为空的 Channel 调用 <code>receive()</code> 方法</li><li>当 <code>isClosedForReceive</code> 为 <code>true</code> 时调用 <code>receive()</code></li></ul>
<p><strong>示例：</strong></p>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateClosedReceiveException() {
    val channel = Channel&lt;String&gt;()
    // 关闭 Channel
    channel.close()
    try {
      // 尝试从已关闭且空的 Channel 接收数据
      val message = channel.receive()
      println("收到消息: $message")
    } catch (e: ClosedReceiveChannelException) {
      println("捕获异常: ${e.message}")
      println("异常类型: ${e::class.simpleName}")
    }
}</pre></div>
<p class="maodian"></p><h3>CancellationException</h3>
<p><strong>触发条件：</strong></p>
<ul><li>Channel 被 <code>cancel()</code> 方法取消</li><li>父协程被取消，导致 Channel 操作被取消</li><li>超时或其他取消信号</li></ul>
<p><strong>示例：</strong></p>
<div class="jb51code"><pre class="brush:java;">suspend fun demonstrateCancellationException() {
    val channel = Channel&lt;String&gt;()
    val job = GlobalScope.launch {
      try {
            // 这个操作会被取消
            channel.send("This will be cancelled")
      } catch (e: CancellationException) {
            println("发送操作被取消: ${e.message}")
            throw e // 重新抛出 CancellationException
      }
    }
    delay(100)
    // 取消 Channel
    channel.cancel(CancellationException("手动取消 Channel"))
    try {
      job.join()
    } catch (e: CancellationException) {
      println("协程被取消: ${e.message}")
    }
}</pre></div>
<p class="maodian"></p><h3>异常与状态关系</h3>
<table><thead><tr><th>Channel 状态</th><th>send() 行为</th><th>receive() 行为</th><th>trySend() 行为</th><th>tryReceive() 行为</th></tr></thead><tbody><tr><td>活跃状态</td><td>正常发送或挂起</td><td>正常接收或挂起</td><td>返回成功/失败结果</td><td>返回成功/失败结果</td></tr><tr><td>发送端关闭</td><td>抛出 ClosedSendChannelException</td><td>正常接收缓冲区数据</td><td>返回失败结果</td><td>正常返回结果</td></tr><tr><td>接收端关闭</td><td>抛出 ClosedSendChannelException</td><td>抛出 ClosedReceiveChannelException</td><td>返回失败结果</td><td>返回失败结果</td></tr><tr><td>已取消</td><td>抛出 CancellationException</td><td>抛出 CancellationException</td><td>返回失败结果</td><td>返回失败结果</td></tr></tbody></table>
<p class="maodian"></p><h3>异常处理技巧</h3>
<p class="maodian"></p><h4>使用非阻塞操作避免异常</h4>
<p>非阻塞操作不会抛出异常，而是返回结果对象：</p>
<div class="jb51code"><pre class="brush:java;">suspend fun safeChannelOperations() {
    val channel = Channel&lt;String&gt;()
    // 安全的发送操作
    val sendResult = channel.trySend("Safe message")
    when {
      sendResult.isSuccess -&gt; println("发送成功")
      sendResult.isFailure -&gt; println("发送失败: ${sendResult.exceptionOrNull()}")
      sendResult.isClosed -&gt; println("Channel 已关闭")
    }
    // 安全的接收操作
    val receiveResult = channel.tryReceive()
    when {
      receiveResult.isSuccess -&gt; println("接收到: ${receiveResult.getOrNull()}")
      receiveResult.isFailure -&gt; println("接收失败: ${receiveResult.exceptionOrNull()}")
      receiveResult.isClosed -&gt; println("Channel 已关闭")
    }
}</pre></div>
<p class="maodian"></p><h4>健壮的异常处理</h4>
<div class="jb51code"><pre class="brush:java;">suspend fun robustChannelUsage() {
    val channel = Channel&lt;String&gt;()
    val producer = GlobalScope.launch {
      try {
            repeat(5) { i -&gt;
                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)
}</pre></div>
<p class="maodian"></p><h2>总结</h2>
<p class="maodian"></p><h3>Channel 关键概念对比</h3>
<table><thead><tr><th>特性</th><th>RENDEZVOUS</th><th>CONFLATED</th><th>BUFFERED</th><th>UNLIMITED</th><th>自定义容量</th></tr></thead><tbody><tr><td><strong>容量</strong></td><td>0</td><td>1</td><td>64</td><td>Int.MAX_VALUE</td><td>指定值</td></tr><tr><td><strong>缓冲行为</strong></td><td>无缓冲，同步</td><td>只保留最新值</td><td>有限缓冲</td><td>无限缓冲</td><td>有限缓冲</td></tr><tr><td><strong>发送阻塞</strong></td><td>是</td><td>否</td><td>缓冲满时</td><td>否</td><td>缓冲满时</td></tr><tr><td><strong>适用场景</strong></td><td>严格同步</td><td>状态更新</td><td>一般异步</td><td>高吞吐量</td><td>批量处理</td></tr><tr><td><strong>内存风险</strong></td><td>低</td><td>低</td><td>中等</td><td>高</td><td>可控</td></tr></tbody></table>
<p class="maodian"></p><h3>溢出策略对比</h3>
<table><thead><tr><th>策略</th><th>行为</th><th>性能特点</th><th>适用场景</th></tr></thead><tbody><tr><td><strong>SUSPEND</strong></td><td>挂起发送操作</td><td>提供背压控制</td><td>确保数据完整性</td></tr><tr><td><strong>DROP_OLDEST</strong></td><td>丢弃旧元素</td><td>发送不阻塞</td><td>实时数据流</td></tr><tr><td><strong>DROP_LATEST</strong></td><td>丢弃新元素</td><td>发送不阻塞</td><td>保护历史数据</td></tr></tbody></table>
<p class="maodian"></p><h3>操作方式</h3>
<table><thead><tr><th>操作类型</th><th>阻塞版本</th><th>非阻塞版本</th><th>异常处理</th><th>返回值</th></tr></thead><tbody><tr><td><strong>发送</strong></td><td><code>send()</code></td><td><code>trySend()</code></td><td>抛出异常</td><td><code>ChannelResult&lt;Unit&gt;</code></td></tr><tr><td><strong>接收</strong></td><td><code>receive()</code></td><td><code>tryReceive()</code></td><td>抛出异常</td><td><code>ChannelResult&lt;T&gt;</code></td></tr><tr><td><strong>特点</strong></td><td>会挂起协程</td><td>立即返回</td><td>需要 try-catch</td><td>通过结果对象判断</td></tr></tbody></table>
<p class="maodian"></p><h3>Channel 状态生命周期</h3>
<table><thead><tr><th>状态</th><th>描述</th><th>send()</th><th>receive()</th><th>检查方法</th></tr></thead><tbody><tr><td><strong>活跃</strong></td><td>正常工作状态</td><td>✅ 正常</td><td>✅ 正常</td><td>-</td></tr><tr><td><strong>发送关闭</strong></td><td>调用 close() 后</td><td>❌ 异常</td><td>✅ 可接收缓冲区数据</td><td><code>isClosedForSend</code></td></tr><tr><td><strong>接收关闭</strong></td><td>缓冲区清空后</td><td>❌ 异常</td><td>❌ 异常</td><td><code>isClosedForReceive</code></td></tr><tr><td><strong>已取消</strong></td><td>调用 cancel() 后</td><td>❌ 异常</td><td>❌ 异常</td><td>-</td></tr></tbody></table>
<p>总体来说，Channel 是一种非常强大的协程通信机制，它可以帮助我们在协程之间进行安全、高效的通信。在使用 Channel时，我们需要注意异常处理、缓冲区容量、溢出策略等问题。</p>
<p>感谢阅读，如果对你有帮助请三连（点赞、收藏、加关注）支持。有任何疑问或建议，欢迎在评论区留言讨论。如需转载，请注明出处：喻志强的博客</p>
<p>到此这篇关于Kotlin 协程之Channel的概念和基本使用详解的文章就介绍到这了,更多相关Kotlin 协程Channel使用内容请搜索琼殿技术社区以前的文章或继续浏览下面的相关文章希望大家以后多多支持琼殿技术社区！</p>
                           
                            <div class="art_xg">
                              <b>您可能感兴趣的文章:</b><ul><li>Kotlin协程Channel源码示例浅析</li><li>Kotlin协程Channel特点及使用细节详解</li><li>Kotlin协程开发之Flow的融合与Channel容量及溢出策略介绍</li><li>Android&nbsp;kotlin中&nbsp;Channel&nbsp;和&nbsp;Flow&nbsp;的区别和选择使用场景分析</li><li>Kotlin&nbsp;Channel处理多个数据组合的流</li></ul>
                            </div>

                        </div>
                        <!--endmain-->
頁: [1]
查看完整版本: Kotlin 协程之Channel的概念和基本使用详解