Tokio 源码深度解析

第13章 channels：mpsc / broadcast / watch / oneshot

作者杨艺韬 · 16,573 字 · 发布于 2026-04-19 · 更新于 2026-09-24

本章要点

Tokio 提供四种 channel：mpsc（多生产者单消费者，最常用）、broadcast（多生产者多消费者，每消息广播）、watch（最新值订阅）、oneshot（一次性发送）
mpsc 的 Chan struct：tx: list::Tx<T>（无锁分块链表）+ rx_waker: AtomicWaker + semaphore: S（背压）——上一章 batch_semaphore 在这里直接复用
oneshot 的惊人简洁：一个 state: AtomicUsize（4 个状态位）+ value: UnsafeCell<Option<T>> + 两个 waker 槽。4 个字段实现一次性发送语义
broadcast 的 ring buffer：buffer: Box<[RwLock<Slot<T>>]>——每 slot 一个 RwLock、所有 receiver 根据自己的 next 位置读
watch = oneshot 但可以反复发送：每次 send 替换当前值 + 递增 version、receiver 比对 version 判断是否有新值
各 channel 的接收侧都是 cancellation-safe——select! 里取消一个 channel 的 recv 不会丢消息（oneshot 的文档没写 Cancel safety 段，这一条是按实现推断；发送侧如 mpsc::Sender::send 被取消会丢掉待发的值）
mpsc 的 rx_waker 是 AtomicWaker（第 7 章讲 LocalSet 时见过）——一个 atomic slot 装一个 Waker，register_by_ref 每次都 clone 一份新 Waker 替换进去（1.40.0 的 AtomicWaker 不做 will_wake 比较，见第 3 章）
核心哲学：channel = 数据结构 + 同步原语。不同 channel 只是"不同数据结构配不同同步策略"

13.1 读本章前的一个框架

channel 是异步编程里最实用的工具——真实项目里几乎每个 Tokio 服务都在用。但 channel 的种类之多（四种 Tokio 原生 + 多种第三方）、语义差异之细（bounded vs unbounded、fair vs unfair、cancel-safe vs not）、让很多人对 channel 的理解停留在"会用 send / recv"的表面。本章要带你做的是"从表面 API 深入到语义 + 实现 + 选型"——读完你选 channel 时不再是拍脑袋、而是基于具体语义做判断。

在深入四种 channel 的实现前，先给你一个统一框架：

Channel = 数据容器 + 生产侧协议 + 消费侧协议

数据容器：链表 / ring buffer / 单值槽
生产侧协议：独占 vs 多生产者；阻塞 vs 非阻塞
消费侧协议：单消费者 vs 多消费者；Polling vs 被 wake

四种 channel 的区别，本质是这三维度不同取值的组合：

channel	数据容器	生产侧	消费侧
mpsc	lock-free 链表	多生产者、bounded 时背压	单消费者、被 AtomicWaker wake
broadcast	环形缓冲（每槽一把 RwLock）	多生产者、互斥尾推进	多消费者、自带的等待链表
watch	单个 `RwLock<T>`	多生产者、替换	多消费者、被 Notify wake
oneshot	单个 `UnsafeCell<Option<T>>`	单生产者、一次	单消费者、状态位通知

带这个框架读本章——每节会明确展开这三个维度。等你读完，所有 Rust 生态里 channel 类的原语（crossbeam、flume、async-channel）你都能用这个框架秒懂。

13.2 四种 channel 的定位

Tokio 提供了四种 channel（mpsc / broadcast / watch / oneshot），不是随意堆砌——它们各自对应一类特定的消息传递模式。mpsc（多生产者单消费者）解决"任务间工作流"、broadcast（多生产者多消费者）解决"事件广播"、watch（多生产者多消费者 + 最新值，Sender 可 clone，见 tokio/src/sync/watch.rs:149）解决"配置分发"、oneshot（一次性单消息）解决"异步结果返回"。这四种模式覆盖了绝大多数真实业务场景——用户几乎不需要自己造 channel、Tokio 的选择足够用。

在深入实现前先厘清四种 channel 的语义差别——选错 channel 是 Tokio 使用里最常见的错误之一。

channel	生产者	消费者	消息消费次数	容量	典型用途
mpsc	多个	1 个	每消息被消费 1 次	bounded / unbounded	任务分发、事件收集
broadcast	多个	多个	每消息每 receiver 消费 1 次	ring buffer（有限）	事件广播、配置变更
watch	多个	多个	读最新值（跳过中间）	1	配置订阅、状态快照
oneshot	1 个	1 个	一次性	1	异步返回值、信号

选 channel 的灵魂拷问：

消息要所有 receiver 都收到吗？是 → broadcast / watch；否 → mpsc / oneshot
消息重要吗（不能丢）？重要 → mpsc / oneshot；可以丢 → broadcast / watch
只发送一次吗？是 → oneshot；多次 → 其他
需要背压（发送端满了等）吗？是 → bounded mpsc；否 → unbounded / broadcast / watch

这 4 个问题 + 上表 ≈ 你所有 channel 选型需求。

channel 的一个哲学问题：Go 还是 Rust 的方式更好

Go 的 chan 极简——一个关键字解决问题。Tokio 的 channel 分 4 类、每类一个模块。哪种哲学更好？

Go 的观点："不用分，一个就够"——简单、易学、一致。 Rust 的观点："不同场景需要不同优化"——性能、类型安全、明确语义。

我的观点：都对，在各自上下文里。

Go 的目标用户是"大多数后端开发者"——优先易用
Rust 的目标用户包括"追求极致性能的系统工程师"——优先灵活和性能

你不能把 Go 的方式强加给 Rust——Rust 本身的定位要求这种细粒度。如果 Tokio 只提供一个 "Channel"，Rust 工程师会不满意（我要 broadcast、我要 watch）。反过来如果 Go runtime 提供 4 种 channel，Go 工程师会嫌复杂。

语言生态的 API 设计反映语言本身的哲学——Tokio 的 4 channel 是 Rust 生态对"细粒度 + 零成本"信仰的诚实表达。读完本章后你不仅理解 Tokio channel，也理解了 Rust 生态的气质。

13.3 mpsc：最常用 channel 的内部

mpsc（multi-producer single-consumer）是所有异步 channel 里最常见的那个——真实 Tokio 项目里的 channel 大多是 mpsc。原因很简单：绝大多数"任务 A 处理完数据交给任务 B"的场景都是这个模式。多个连接任务把请求交给同一个处理任务、后台任务把结果写到日志线程——都是 mpsc（多个 worker 共同消费一条队列则不是，见 13.8 末尾的 task queue 案例）。理解 mpsc 的内部实现几乎等于理解异步 channel 的核心——本节讲的是 Tokio 最精巧的代码之一。

mpsc 是 Tokio 用得最多的 channel——几乎所有"多 Task 向一个 Task 发消息"的场景都用它。打开 tokio/src/sync/mpsc/chan.rs。原样（省略了字段上的文档注释）：

// 来源：tokio-rs/tokio · tokio/src/sync/mpsc/chan.rs:52 (tokio-1.40.0)
pub(super) struct Chan<T, S> {
    tx: CachePadded<list::Tx<T>>,
    rx_waker: CachePadded<AtomicWaker>,
    notify_rx_closed: Notify,
    semaphore: S,
    tx_count: AtomicUsize,
    tx_weak_count: AtomicUsize,
    rx_fields: UnsafeCell<RxFields<T>>,
}

7 个字段，每个都有明确角色：

tx: list::Tx<T>（tokio/src/sync/mpsc/chan.rs:54）—— 发送端持有的 "lock-free 链表"尾端句柄（block_tail + tail_position）。发送就是往这个链表尾部 push
rx_waker: AtomicWaker（tokio/src/sync/mpsc/chan.rs:57）—— 接收端等待时存的 Waker。发送端 push 完后 wake 它
notify_rx_closed: Notify（tokio/src/sync/mpsc/chan.rs:60）—— 发送端等"接收端关闭"用的（Sender::closed() 在它上面等，接收端关闭时 notify_waiters）
semaphore: S —— 泛型的「容量策略」。bounded 版内含上一章的 batch_semaphore::Semaphore 加一个 bound；unbounded 版就是一个 AtomicUsize 计数器
tx_count: AtomicUsize（tokio/src/sync/mpsc/chan.rs:68）—— 当前活跃 sender 数。归零时 channel 自动关闭
tx_weak_count: AtomicUsize（tokio/src/sync/mpsc/chan.rs:71）—— WeakSender 的数量（不阻止 channel 关闭）
rx_fields: UnsafeCell<RxFields<T>>（tokio/src/sync/mpsc/chan.rs:74）—— 接收端私有数据（用 UnsafeCell 因为接收端独占、不需要锁）

注意两处 CachePadded——把 tx 和 rx_waker 分别填充到独立的 cache line、防止 false sharing（第 6 章讲过）。mpsc 的 hot path 上 sender 反复改 tx、receiver 反复改 rx_waker——分到不同 cache line，可以减少两者互相把对方的缓存行打失效。

`Tx` 和 `Rx`：对 Chan 的包装

// 来源：tokio/src/sync/mpsc/chan.rs:19 与 tokio/src/sync/mpsc/chan.rs:30
pub(crate) struct Tx<T, S> {
    inner: Arc<Chan<T, S>>,
}

pub(crate) struct Rx<T, S: Semaphore> {
    inner: Arc<Chan<T, S>>,
}

两个都是 Arc<Chan<T, S>> 的薄包装。Tx 可以 clone（多个 sender），Rx 不能 clone（单一 receiver——这就是"mpsc 的 SC"——Single Consumer）。

send 和 recv 的核心流程

send（bounded）：

semaphore.acquire(1).await —— 拿一个 permit（满了就等）
list::Tx::push(value) —— 无锁 push 到链表尾
rx_waker.wake() —— 唤醒等 recv 的 Task

recv：

尝试 list::Rx::pop() —— 拿到就返回 Ready
空：注册 Waker 到 rx_waker
双检查 list::Rx::pop() —— 期间可能有人 push（第 8 章讲过的 register-and-check 模式）
仍然空 → 返回 Pending

这条 send/recv 路径用了上一章的 Semaphore（背压）+ 第 7 章见过的 AtomicWaker（唤醒）+ lock-free 链表（无锁队列）——三件法宝组合出一个 mpsc channel。

`AtomicWaker` 和 `Notify` 的角色分工

Chan 里有两个唤醒机制：

rx_waker: AtomicWaker：接收端唯一的 Waker 槽。只有一个 receiver 嘛，一个 slot 够
notify_rx_closed: Notify：多个 sender 可能等"receiver 关闭"的通知——需要 multi-waiter 的 Notify

用对应的原语——一个等待者用 AtomicWaker、多个等待者用 Notify。原语的选择反映等待关系。

`tx_count` 和 `tx_weak_count`

Chan 里有两个计数：

tx_count：强引用计数，归零触发 channel 关闭
tx_weak_count：弱引用计数，不阻止关闭

WeakSender 的用途：有些模块只想"如果 receiver 还在，我偶尔发消息"——不想因为自己持有 sender 而让 receiver 误以为"还有活跃 sender"。WeakSender 就是为这个场景设计。

典型案例：监控模块。服务主流程是 request-response，监控只是旁观者。监控持 WeakSender 发 metrics——不会因为监控持有 sender 而阻止 channel 关闭。

13.4 mpsc 的 lock-free 链表

Tokio mpsc 的底层数据结构是无锁的分块链表——多个生产者用一次 fetch_add 抢一个槽位序号、单个消费者按序号顺序读。它不是教科书上那种「一个节点一个 Box」的链表：每个 Block 里放一段连续数组（tokio/src/sync/mpsc/mod.rs:134 起连着三个 cfg：64 位平台 BLOCK_CAP = 32、非 64 位是 16、loom 测试下调成 2），块与块之间用 AtomicPtr 串起来。好处是攒满一整块才分配一次内存（64 位机上就是每 32 条消息一次）。

同一块内的消息在内存里连续——tokio/src/sync/mpsc/block.rs:18 的注释原话是 "continuous array in order to improve cache line behavior when reading"。再加上 Chan 里 tx 和 rx_waker 各自 CachePadded，生产端和消费端的热点字段不会落在同一条 cache line 上。

list::Tx<T>（tokio/src/sync/mpsc/list.rs:12）/ list::Rx<T>（tokio/src/sync/mpsc/list.rs:22）是 Tokio 自己实现的无锁队列——专门为 mpsc 场景优化。原样：

// 来源：tokio/src/sync/mpsc/list.rs:12 与 tokio/src/sync/mpsc/list.rs:22
pub(crate) struct Tx<T> {
    /// Tail in the `Block` mpmc list.
    block_tail: AtomicPtr<Block<T>>,

    /// Position to push the next message. This references a block and offset
    /// into the block.
    tail_position: AtomicUsize,
}

/// List queue receive handle
pub(crate) struct Rx<T> {
    /// Pointer to the block being processed.
    head: NonNull<Block<T>>,
    /// Next slot index to process.
    index: usize,
    /// Pointer to the next block pending release.
    free_head: NonNull<Block<T>>,
}

// 简化自 tokio/src/sync/mpsc/block.rs:13（注释为节选改写）
pub(crate) struct Block<T> {
    header: BlockHeader<T>,
    /// Values pushed into this block, stored in a continuous array.
    values: Values<T>,   // 内部是 [UnsafeCell<MaybeUninit<T>>; BLOCK_CAP]
}

注意 Tx 全是原子字段、Rx 全是普通字段——多个 sender 并发推进 tail，单个 receiver 独占 head，两侧的保护强度按访问模式区分。

为什么专门做 lock-free 链表而不用 VecDeque？

VecDeque 增删需要 Mutex 保护 → 锁竞争
lock-free 链表 用原子操作（抢槽用 fetch_add、追加新块用 CAS）让 push/pop 无锁 → hot path 零锁

push 短得出人意料（原样，tokio/src/sync/mpsc/list.rs:68）：

// 来源：tokio/src/sync/mpsc/list.rs:68
pub(crate) fn push(&self, value: T) {
    // First, claim a slot for the value. `Acquire` is used here to
    // synchronize with the `fetch_add` in `reclaim_blocks`.
    let slot_index = self.tail_position.fetch_add(1, Acquire);

    // Load the current block and write the value
    let block = self.find_block(slot_index);

    unsafe {
        // Write the value to the block
        block.as_ref().write(slot_index, value);
    }
}

三步、没有 CAS 重试循环：

tail_position.fetch_add(1, Acquire) —— 一次无条件的原子自增就把槽位抢到手。多个 sender 并发调用时每人拿到的 slot_index 天然互不相同，不存在"CAS 失败要重来"
find_block(slot_index)（tokio/src/sync/mpsc/list.rs:96）—— 由序号算出它属于哪个块、块内第几槽（start_index 和 offset 本质就是对 BLOCK_CAP 做除法和取模，而 BLOCK_CAP 被要求是 2 的幂——tokio/src/sync/mpsc/block.rs:52 的 BLOCK_MASK 和 :55 的 SLOT_MASK 直接把它变成位运算）；需要新块时在这里追加
block.write(slot_index, value) —— 把值写进那个槽，并置上"此槽已就绪"的标志

"先 fetch_add 抢序号、再按序号定位存储位置"是无锁队列里最省心的写法——它把"多写者竞争"压缩成一条原子指令。真正的复杂度在 find_block 的块追加、以及消费端读完后如何安全回收块（reclaim_blocks，tokio/src/sync/mpsc/list.rs:332）——tokio/src/sync/mpsc/list.rs 的 398 行加 tokio/src/sync/mpsc/block.rs 的 461 行一共 859 行，绝大部分都在处理这些边界。

为什么 mpsc 不用 ring buffer

你可能想：用 ring buffer（像 broadcast）不是更省内存吗？

因为 mpsc 要支持 unbounded——ring buffer 有固定大小，unbounded 场景要么不行、要么动态扩容（复杂且慢）。链表天然无限制。

此外 mpsc 的接收端只有一个——链表消费端无竞争。broadcast 需要 N 个 receiver 并发读——ring buffer + per-slot RwLock 才能处理。

每种 channel 的底层数据结构是针对它语义选的最优——不是随意选的。链表 for mpsc / ring for broadcast / single slot for oneshot+watch —— 都有深层原因。

list::Rx 和 list::Tx 的分工

注意 Chan 里 tx: list::Tx<T> ——只有 Tx、没有 Rx。这是因为 list::Rx 的状态是 rx 私有的（存在 rx_fields: UnsafeCell<RxFields<T>> 里）。

为什么要分 Tx / Rx？

list::Tx 的字段被多个 sender 并发写——需要 atomic
list::Rx 的字段只被单一 receiver 独占访问——不需要 atomic、用 UnsafeCell 就够

这种"按访问模式拆分数据结构"又是 Tokio 的签名模式（见第 5 章的 Core / Shared、第 8 章的 Driver / Handle）。不要"一个大 struct 所有字段都 atomic" —— 识别出每个字段的实际访问模式，分别保护。

unbounded 不带 Semaphore

Chan<T, S> 的 S 是一个 crate 内部的 Semaphore trait（不是 sync::Semaphore 那个公开类型），两种 mpsc 各给它一个实现：

// 来源：tokio/src/sync/mpsc/bounded.rs:173
pub(crate) struct Semaphore {
    pub(crate) semaphore: semaphore::Semaphore,   // 就是上一章的 batch_semaphore
    pub(crate) bound: usize,
}

// 来源：tokio/src/sync/mpsc/unbounded.rs:106
pub(crate) struct Semaphore(pub(crate) AtomicUsize);

unbounded 版不是空壳、也不是 ZST——它是一个裸 AtomicUsize，只用来数消息条数（最低位仍留给 closed 标志，所以 impl Semaphore for unbounded::Semaphore 里 add_permit 写的是 fetch_sub(2)（tokio/src/sync/mpsc/chan.rs:572）、add_permits(n) 是 fetch_sub(n << 1)（tokio/src/sync/mpsc/chan.rs:581））。它不阻塞任何人，但 is_closed() 以及接收端判断"还有没有在途消息"的 is_idle() 都要靠它（len() 则是用链表的 tail_position 减去读位置算出来的，tokio/src/sync/mpsc/list.rs:253）。

bounded 用真信号量、unbounded 用一个计数器，共用同一套 Chan 代码——这是 trait 抽象的经典应用：同一份队列逻辑，按泛型参数换掉"容量策略"这一块。

13.5 oneshot：一个 atomic 搞定一切

oneshot 是 Tokio channel 里最小的那个——它只能发送一次消息、接收一次、然后结束。虽然功能简单，但它的使用频率极高：任何"异步等一个结果"的场景都用得上 oneshot（注意 JoinHandle 和 timeout 并不用它——前者走 Task 自己的状态字和输出槽，后者就是 Sleep 加被包的 Future）。因为只需要一次通信、实现可以极致简化——一个 AtomicUsize 就能存下整个状态机。这是 Tokio "用最小原语实现最高频场景"哲学的完美体现。

oneshot 是 "一次性发送" channel——sender 发一个值，receiver 收一个值。用得最多的场景是 async fn 的返回值——spawn 一个 Task、用 oneshot 拿它的结果。

Tokio oneshot 的 Inner 结构（简化）：

// 来源：tokio/src/sync/oneshot.rs:379（`Task` 在 tokio/src/sync/oneshot.rs:404）
struct Inner<T> {
    /// Manages the state of the inner cell.
    state: AtomicUsize,
    /// The value. This is set by `Sender` and read by `Receiver`. The state of
    /// the cell is tracked by `state`.
    value: UnsafeCell<Option<T>>,
    /// The task to notify when the receiver drops without consuming the value.
    ///
    /// ## Safety
    ///
    /// The `TX_TASK_SET` bit in the `state` field is set if this field is
    /// initialized. If that bit is unset, this field may be uninitialized.
    tx_task: Task,
    /// The task to notify when the value is sent.
    ///
    /// ## Safety
    ///
    /// The `RX_TASK_SET` bit in the `state` field is set if this field is
    /// initialized. If that bit is unset, this field may be uninitialized.
    rx_task: Task,
}

struct Task(UnsafeCell<MaybeUninit<Waker>>);

4 个字段 —— 核心是 state: AtomicUsize，塞进了 4 个状态位（原样，tokio/src/sync/oneshot.rs:1287-tokio/src/sync/oneshot.rs:1304，注释为本专栏所加）：

const RX_TASK_SET: usize = 0b00001;   // receiver 的 waker 已写入 rx_task
const VALUE_SENT:  usize = 0b00010;   // 值已存入 value 字段
const CLOSED:      usize = 0b00100;   // receiver 关闭了 channel
const TX_TASK_SET: usize = 0b01000;   // sender 的 waker 已写入 tx_task

注意两个 waker 槽是 UnsafeCell<MaybeUninit<Waker>> 而不是 Option<Waker>——这不是为了省空间（Waker 里的 vtable 引用非空，Option<Waker> 本来就和 Waker 一样大，64 位上都是 16 字节），而是"有没有 waker"已经由 state 里对应的 *_TASK_SET 位记账，不必在字段里再存一份；代价是"这块内存有没有初始化"完全靠这几个位来保证。源码在两处字段上都写了 ## Safety 注释来说明这个不变式。

又是一个 bit packing 的 AtomicUsize——你在本专栏里见过这个模式很多次了（Task state、IO readiness、Timer 的 cached_when、semaphore 的 CLOSED 位、watch 的版本号）。

send 和 poll 的协调

send（简化自源码，去掉大段 SAFETY 注释）：

// 示意，非逐字：tokio/src/sync/oneshot.rs:595（`complete` 在 tokio/src/sync/oneshot.rs:1125，`set_complete` 在 tokio/src/sync/oneshot.rs:1315）
pub fn send(mut self, t: T) -> Result<(), T> {
    let inner = self.inner.take().unwrap();

    // 1. 先把值写进 UnsafeCell —— 此时 VALUE_SENT 还没置位，接收端不会来读
    inner.value.with_mut(|ptr| unsafe { *ptr = Some(t); });

    // 2. complete() 里用 CAS 循环置 VALUE_SENT（已 CLOSED 就不置），并拿回旧状态
    if !inner.complete() {
        // 3. 旧状态里有 CLOSED —— 接收端已经不要了，把值取回来还给调用方
        return Err(unsafe { inner.consume_value().unwrap() });
    }

    Ok(())
}

fn complete(&self) -> bool {
    let prev = State::set_complete(&self.state);   // compare_exchange_weak 循环：未 CLOSED 才置 VALUE_SENT

    if prev.is_closed() {
        return false;
    }

    if prev.is_rx_task_set() {
        unsafe { self.rx_task.with_task(Waker::wake_by_ref); }
    }

    true
}

核心：先写值、再用一次 CAS 原子地"发布 + 读回旧状态"（set_complete 特意没用 fetch_or：源码注释说明，通道已关闭时不能再置 VALUE_SENT，否则发送端取回值和接收端读值会同时碰那个 UnsafeCell）——旧状态同时回答了两个问题："接收端关了吗"和"接收端的 waker 挂上了吗"，据此决定 wake、返回 Err 还是返回 Ok。

写值和置位的顺序不能反——源码的 SAFETY 注释写得很明白：接收端只在看到 VALUE_SENT 之后才会去碰那个 UnsafeCell，所以"值先落地、标志后置"才不会读到半初始化的内存。

poll（receiver 侧）：

// 示意，非逐字：真实实现是 tokio/src/sync/oneshot.rs:1142 的 Inner::poll_recv，
// 另有 will_wake 复用旧 waker、CLOSED 分支、Sender 未发送就 drop 时返回 RecvError、coop 预算等处理
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<T, RecvError>> {
    let state = self.inner.state.load(Acquire);
    
    if state & VALUE_SENT != 0 {
        // 值已经到了
        let value = unsafe { (*self.inner.value.get()).take().unwrap() };
        return Poll::Ready(Ok(value));
    }
    
    // 值没到——注册 waker（写进 MaybeUninit<Waker> 槽）
    unsafe { self.inner.rx_task.set_task(cx); }
    
    // 设置 RX_TASK_SET 位、双检查 VALUE_SENT
    let state = self.inner.state.fetch_or(RX_TASK_SET, AcqRel);
    if state & VALUE_SENT != 0 {
        // 刚好在我们注册的瞬间值到了——直接拿
        let value = unsafe { (*self.inner.value.get()).take().unwrap() };
        return Poll::Ready(Ok(value));
    }
    
    Poll::Pending
}

这就是第 8 章讲的 "register + double-check" 模式——避免注册 Waker 和值到达的 race。

整个 oneshot 的代码极少——Inner 只有 4 字段、逻辑清晰。它是"最小并发原语"的典范。

oneshot 的开销构成

一次 oneshot 通信要付出的成本可以完全数清楚（本机没有可复现的基准环境，这里只列构成）：

一次 Arc<Inner<T>> 的堆分配（channel 创建时）
发送侧：一次写 UnsafeCell，一次 CAS 置 VALUE_SENT（有竞争时重试），可能加一次 wake_by_ref()
接收侧：一次 load，或者"写 waker + fetch_or(RX_TASK_SET)"（fetch_or 返回的旧状态就是复查）
销毁：两个 Arc 各一次 fetch_sub

总共是个位数的原子操作 + 一次堆分配——没有锁、没有队列、没有链表。这就是为什么"异步等一个结果"这件事在 Tokio 里几乎不用考虑成本。JoinHandle 内部用的是同一类思路：一个原子状态字 + 一个值槽（第 6 章讲过）。

oneshot 的"Receiver.close()" 技巧

Oneshot receiver 有一个特殊方法 close()：

// 来源：tokio/src/sync/oneshot.rs:1212（Inner::close）
fn close(&self) -> State {
    let prev = State::set_closed(&self.state);   // fetch_or(CLOSED)

    if prev.is_tx_task_set() && !prev.is_complete() {
        unsafe { self.tx_task.with_task(Waker::wake_by_ref); }
    }

    prev
}

注意那个 && !prev.is_complete()——值已经发出来了就没必要再去叫醒发送端。

用途：receiver 提前决定不要结果了——告诉 sender 别再费事发。典型场景：

let (tx, mut rx) = oneshot::channel::<BigData>();

spawn(async move {
    let data = expensive_compute().await;
    let _ = tx.send(data);  // 如果 receiver 关了，send 返回 Err(data)——sender 知道不用 send
});

tokio::select! {
    result = &mut rx => { use_result(result); }
    _ = timeout => {
        rx.close();  // 标记关闭：之后 tx.send 返回 Err；sender 要提前收手得自己查 tx.is_closed() 或等 tx.closed()
    }
}

这个机制让 oneshot 比 Go channel 更优雅——Go channel 的 receiver 关闭不能反向通知 sender。Rust 在这里用 Option 传递 + 显式 close —— 细节做得更周到。

oneshot 不能 Clone Sender

let (tx1, rx) = oneshot::channel();
let tx2 = tx1.clone();  // ❌ 编译错误：Sender 没有 Clone impl

设计上故意。oneshot 的语义是"一次性发送"——两个 sender 同时 send 就语义不清了。Rust 通过不实现 Clone 把这个约束编码进类型。

对比 mpsc：Sender::clone() 可以，因为 mpsc 明确允许多 sender。

API 和语义对齐是 Tokio 设计的深处原则——别提供语义上不该支持的操作。

oneshot 真实开销数字

把 oneshot 和 mpsc 摆在一起看，成本差别一目了然（结构上的差别，不是某台机器上的数字）：

	oneshot	mpsc
创建	一次 `Arc<Inner<T>>` 分配	一次 `Arc<Chan>` 分配 + 首个 `Block`
发送	写一次 `UnsafeCell` + 一次 CAS 置位	信号量 acquire + `fetch_add` 抢槽 + 可能追加新块 + 唤醒
接收	一次 `load`（或注册 waker 后复查）	读块内槽位 + 推进 index + 可能回收块
同步原语	无	背压用 batch_semaphore、唤醒用 AtomicWaker

oneshot 之所以便宜，是因为它连"队列"都没有——一个值槽、一个状态字、两个 waker 位置，仅此而已。

这也是为什么 JoinHandle.await 拿 Task 返回值几乎不花什么代价——它走的是同一类结构（第 6 章讲过）。

oneshot 为什么不是 Copy

Sender::send(self, value: T) 消耗 self——这要以 Sender 不能 Copy 为前提（否则按值传进去的只是一份拷贝，原件还能再 send）。为什么不让它 Copy？

因为"一次性语义"。如果 Sender 可 Copy，两个 copy 都可以 send——但内部只有一个 value slot。类型系统禁止这件事是正确的。

整个专栏里 Tokio 多次通过"自定义类型 + 不实现 Copy/Clone"把业务规则编码进类型系统。这种"让编译器帮我检查业务规则"的思路是 Rust 的特色——很多其他语言只能靠运行时断言或文档约定。

13.6 broadcast：ring buffer + per-slot RwLock

broadcast channel 面临一个独特的挑战：如果 receiver 消费慢、是应该阻塞 sender、还是丢弃老消息？。Tokio 选了后者——broadcast 是有损的、receiver 太慢就跳过未读的老消息、收到 RecvError::Lagged 错误。这种"速度优先、丢弃慢者"的设计让 broadcast 适合"事件广播"场景（日志、告警、状态变更）——不适合"必须每条都收到"的场景（那种场景给每个消费者各开一条 bounded mpsc，由发送端逐个投递）。

broadcast 是唯一多生产者多消费者的 channel。每个消息所有活跃 receiver 都能收到。底层不是链表——是 ring buffer。

// 来源：tokio/src/sync/broadcast.rs:304（`Slot` 在 tokio/src/sync/broadcast.rs:334）
struct Shared<T> {
    /// slots in the channel.
    buffer: Box<[RwLock<Slot<T>>]>,
    /// Mask a position -> index.
    mask: usize,
    /// Tail of the queue. Includes the rx wait list.
    tail: Mutex<Tail>,
    /// Number of outstanding Sender handles.
    num_tx: AtomicUsize,
}

struct Slot<T> {
    /// Remaining number of receivers that are expected to see this value.
    /// When this goes to zero, the value is released.
    rem: AtomicUsize,
    /// Uniquely identifies the `send` stored in the slot.
    pos: u64,
    /// The value being broadcast.
    val: UnsafeCell<Option<T>>,
}

4 个字段：

buffer: Box<[RwLock<Slot<T>>]>（tokio/src/sync/broadcast.rs:306）—— 固定大小的数组，每槽一把 RwLock（多 receiver 并发读、写者独占写）
mask: usize（tokio/src/sync/broadcast.rs:309）—— capacity - 1，用 pos & mask 取模（capacity 被强制成 2 的幂）
tail: Mutex<Tail>（tokio/src/sync/broadcast.rs:312）—— 写者用的尾指针；注意源码注释写着 "Includes the rx wait list"——接收者的等待链表就挂在 Tail 里，broadcast 既不用 Semaphore 也不用 Notify，是自己一套
num_tx: AtomicUsize（tokio/src/sync/broadcast.rs:315）—— sender 数量

Slot::rem 是理解 broadcast 的钥匙：它记着"还有几个接收者没看过这条消息"。每个接收者读完把它减 1，减到 0 时值才被 drop（tokio/src/sync/broadcast.rs:1496 的 RecvGuard::drop 里 if 1 == self.slot.rem.fetch_sub(1, SeqCst) 就把槽位置回 None）。注意没有"最后一个读者拿走原值"的优化——取值一律走 RecvGuard::clone_value（tokio/src/sync/broadcast.rs:1485），每个接收者都拿一份克隆。

Receiver（tokio/src/sync/broadcast.rs:204）：

// 来源：tokio/src/sync/broadcast.rs:204
pub struct Receiver<T> {
    shared: Arc<Shared<T>>,
    next: u64,       // 下一个要读的 slot position
}

每 receiver 独立持一个 next: u64——这就是为什么 receiver 不能 clone（每个 receiver 有自己的位置）。

写流程（`send`）

拿 tail: Mutex<Tail> 锁；tail.rx_cnt == 0 直接返回 Err(SendError(value))
算 slot = tail.pos & mask，并先把 tail.pos 加一
拿 buffer[slot] 的 RwLock 写锁（独占）
写 slot.pos、把 slot.rem 置成当前接收者数、把 value 放进 Slot
先 drop(slot) 放掉槽锁，再 notify_rx(tail) 唤醒等待者并释放 tail 锁——源码注释特意说明这个顺序不能反（send 在 tokio/src/sync/broadcast.rs:586，这段注释在 tokio/src/sync/broadcast.rs:613-tokio/src/sync/broadcast.rs:619）

写竞争由 tail 的 Mutex 串行化——所有 sender 串行写。这是 broadcast 比 mpsc 慢的主要原因（mpsc 的 sender 基本 lock-free）。

读流程（`recv`）

算 slot = self.next & mask
拿 buffer[slot] 的 RwLock 读锁（共享）
比对 slot.pos == self.next —— 相等就是有新消息可读；不等才放掉槽锁、去拿 tail 锁，区分"暂时没新消息（挂 waker）"和"已被覆盖（Lagged）"（recv_ref，tokio/src/sync/broadcast.rs:1053）
增加 self.next
clone 出 value（每 receiver 收到自己的 clone）

注意"clone value"——broadcast 的代价：消息必须 T: Clone。不像 mpsc 每消息只流一份、broadcast 每个 receiver 各一份 clone。

broadcast 的 Clone 要求

broadcast::channel::<T>(n) 要求 T: Clone——因为每个 receiver 收到的是 value 的 clone。

如果 T 不实现 Clone（比如 File、TcpStream 等独占资源），broadcast 用不了——这类资源天然不能"广播"。

如果你真的需要广播独占资源的"访问权"，考虑：

Arc<T>：多个 receiver 共享访问
Arc<Mutex<T>>：多 receiver 轮流独占访问
oneshot x N：每 receiver 一个 oneshot、发送端持有多个 oneshot sender

每种方案都有自己的语义 trade-off——选最符合你场景的。

receiver 的 `resubscribe`

Receiver 可以 resubscribe() 创建一个新 receiver，起始位置是当前 tail（即从这一刻起开始接收、之前的消息不收）。

let mut rx2 = rx.resubscribe();
// rx2 只会收到 resubscribe 之后 send 的消息

适合 "从 tail 开始订阅" 场景——new subscriber 不需要历史消息、只要新消息。和 tx.subscribe() 行为相同、但从 receiver 调用更自然（无需持有 sender）。

`capacity` 方法查当前空闲

bounded mpsc 的 Sender / Receiver 都有 capacity()（当前空闲 permit 数）/ max_capacity() / len()；broadcast 只有 len()，没有 capacity()。生产用途：监控、诊断 "channel 是否满了"。

// rx 是 bounded mpsc::Receiver
if rx.capacity() < 10 {
    warn!("channel near capacity, possible backpressure");
}

别在 hot path 用——这些方法有原子操作开销（broadcast 的 len() 还要拿 tail 锁）、频繁调会降低吞吐。偶尔诊断 OK。

Lagged：receiver 跟不上怎么办

如果某个 receiver 太慢，tail.pos 超前它 capacity 个 slot——它的 next 对应的 slot 已经被覆盖。这时 recv 返回 Err(Lagged(skipped_count))——告诉 receiver "你已经落后 N 个消息"。

receiver 可以选择：

接受丢失：继续 recv——游标已被挪到缓冲里仍保留的最旧一条，从那里接着读
放弃：关闭 receiver 结束

broadcast 的设计：快 sender 优先，慢 receiver 自己承担 lag 后果。这防止一个慢 receiver 拖垮整个 channel 的吞吐——经典的 "slow consumer" 问题的解决策略。

broadcast 的容量必须 2 的幂

channel::<T>(capacity) 时 capacity 会被自动 round 到下一个 2 的幂：

let (tx, rx) = broadcast::channel(10);   // 实际容量 16（round up to 16）

为什么：mask = capacity - 1，算 slot 用 pos & mask 取模——要求 capacity 是 2 的幂才能把取模变成位与。源码里就一行 capacity = capacity.next_power_of_two();，注释写着 "Round to a power of two"。

传非 2 的幂不会 panic、也不警告——透明地向上取整。真正会 panic 的是两种极端：capacity == 0（"broadcast channel capacity cannot be zero"）和 capacity > usize::MAX >> 1。

broadcast 的 capacity 和内存开销

关键一点：buffer 是在 channel(capacity) 时一次性分配好的，不是按需增长——

// 来源：tokio/src/sync/broadcast.rs:510
let mut buffer = Vec::with_capacity(capacity);
for i in 0..capacity {
    buffer.push(RwLock::new(Slot { /* rem, pos, val */ }));
}

每个槽是一个 RwLock<Slot<T>>，Slot<T> 里有 AtomicUsize + u64 + UnsafeCell<Option<T>>。具体多少字节取决于 T 和平台，但形状是确定的：创建即占用 capacity 份槽位空间，哪怕一条消息都没发。

建议：broadcast 用在广泛订阅 场景（少数 channel、多订阅者），不适合 "每实体一个 broadcast"（成千上万个 channel、每个只有少数订阅者）——后者用 mpsc 或 watch 更合适。

broadcast 为什么比 mpsc 贵

不用测也能从结构上看出来 broadcast 的每一步都比 mpsc 重：

步骤	mpsc	broadcast
发送端互斥	只有一次 `fetch_add` 抢序号，无锁	要拿 `tail: Mutex<Tail>`，所有 sender 串行
写入	直接写块内槽位	还要拿 `buffer[slot]` 的 `RwLock` 写锁
唤醒	一个 `AtomicWaker`	遍历 `Tail` 里的等待链表唤醒所有接收者
接收	移出值，零拷贝	拿 `RwLock` 读锁 + `clone()` 一份值
回收	整块回收	每个 `Slot` 有 `rem: AtomicUsize` 计数，减到 0 才 drop 值

结论是定性的、也是可靠的：如果你不需要"每条消息每个订阅者都收到"这个语义，就别用 broadcast。

内存也别忽略：容量会被向上取整到 2 的幂，每个槽都是一个 RwLock<Slot<T>>（内含 AtomicUsize + u64 + Option<T>），在创建时就全部分配好。所以 broadcast 适合"少数几个 channel、很多订阅者"，而不是"每个实体一个 broadcast"——后者用 mpsc 或 watch 更合适。

13.7 watch：最新值订阅

watch channel 的语义非常有意思——它不保证每条消息都被收到、但保证读到的永远是最新值。这个语义看似奇怪、但在真实系统里用途很多：配置热更新（服务只关心当前配置、不关心中间经过几次修改）、状态广播（比如连接状态、服务健康状态、最新价格）、统计快照（指标值、队列深度、CPU 使用率）。watch 的 "跳过中间态" 是特性不是缺陷——它让订阅者永远看到最新快照、不会因为中间过时值浪费处理时间。

watch 适合"订阅最新配置 / 状态"——receiver 只关心当前值，不关心历史。

Shared 结构（简化）：

// 来源：tokio/src/sync/watch.rs:244
struct Shared<T> {
    /// The most recent value.
    value: RwLock<T>,

    /// The current version.
    ///
    /// The lowest bit represents a "closed" state. The rest of the bits
    /// represent the current version.
    state: AtomicState,

    /// Tracks the number of `Receiver` instances.
    ref_count_rx: AtomicUsize,
    /// Tracks the number of `Sender` instances.
    ref_count_tx: AtomicUsize,

    /// Notifies waiting receivers that the value changed.
    notify_rx: big_notify::BigNotify,
    /// Notifies any task listening for `Receiver` dropped events.
    notify_tx: Notify,
}

核心是 state（tokio/src/sync/watch.rs:252）里的 version：每次 send 替换 value、version 加一。receiver 记自己上次见到的 version、poll 时对比。注意 state 的最低位又被借去当 closed 标志——和 batch_semaphore 的 CLOSED、oneshot 的状态位是同一个套路。

唤醒接收者用的是 BigNotify（tokio/src/sync/watch.rs:261 的字段，类型定义在 tokio/src/sync/watch.rs:335）而不是裸 Notify——它内部持有 8 个 Notify，接收者等待时随机（没有随机数源时轮流）挑一个排队，notify_waiters 则把 8 个逐一通知；源码注释说目的是避免大量接收者争同一个 Notify 内部的锁。

send：

// 示意，非逐字：简化自 tokio/src/sync/watch.rs（send :977 → send_replace → send_modify → send_if_modified :1084）
// 真实 send 在没有 receiver 时直接返回 Err(SendError(value))
fn send(&self, value: T) {
    let mut lock = self.inner.value.write().unwrap();
    *lock = value;
    self.inner.state.fetch_add(STEP_SIZE, Release);   // 持写锁期间递增版本
    drop(lock);
    self.inner.notify_rx.notify_waiters();
}

recv (changed().await)：

// 示意，非逐字：真实实现是 tokio/src/sync/watch.rs:896 的 changed_impl
fn changed(&mut self) -> ... {
    loop {
        // 先登记通知、再查版本——反过来会漏掉两步之间的 send
        let notified = self.shared.notify_rx.notified();
        let state = self.shared.state.load(Acquire);
        let current_version = version_from(state);
        if current_version != self.seen_version {
            self.seen_version = current_version;
            return Ok(());
        }
        // 没变化——等 notify
        notified.await;
    }
}

watch vs broadcast 的关键差异：

broadcast：每个消息被每个 receiver 独立消费，"不丢历史"
watch：只保留最新值，如果 sender 连续 send 10 次、receiver 只会感知"有变化"、看到最新那个

配置热更新的完美 fit：你只想知道"现在配置是什么"，不在乎中间变了多少次。省了 broadcast 的 ring buffer 开销。

watch 的一个高级 API：`wait_for`

// 等到满足 predicate 的值
let value: Ref<'_, T> = rx.wait_for(|v| v.is_ready()).await?;

签名是（原样）：

// 来源：tokio/src/sync/watch.rs:809
pub async fn wait_for(
    &mut self,
    mut f: impl FnMut(&T) -> bool,
) -> Result<Ref<'_, T>, error::RecvError>

注意它返回的是 Ref<'_, T> 而不是 T——一个借用守卫，所以 T 不需要 Clone。内部逻辑是"拿读锁 → 顺手把自己的 version 同步到最新 → 用 f 判断 → 不满足就 changed().await 再来一轮"。

适合"等配置达到某个状态"——比起手写循环更简洁、语义明确。

Tokio 在这类"封装常见循环"的工作上很有节制——不会为每个模式都加 helper，只加 wait_for 这类够通用 + 够常见的。

watch 的 `initial` 值语义

watch::channel(initial) 要求一个初始值——这和 mpsc / broadcast 不同（它们空初始）。为什么？

因为 watch 的语义是"当前值"——必须有值。如果 "刚创建就是空"，borrow() / changed() 语义就复杂——你还没 "当前值" 呢。

这个小约束让 API 语义极简：receiver 从一开始就能 borrow、第一次 changed 要等真的 send。

对用户的影响：有时候你没有"合理初始值"，要编一个或用 Option<T> 包装——小开销但换来 API 清晰。

watch 的 `borrow()` vs `borrow_and_update()`

watch 的 receiver 有两个读方法：

borrow()：读当前值、不更新 seen_version。下次 changed().await 仍然会立刻返回（因为它还没"承认"自己看过）
borrow_and_update()：读当前值、更新 seen_version。下次 changed().await 要等真的有新值

微妙但重要。经验法则：

如果你想"只要当前值"（比如偶尔查一次配置）→ borrow()
如果你想"处理完这个版本再等下一个"→ borrow_and_update()

混用会导致 changed() 循环逻辑错乱。

watch 不保留历史的代价

watch 的设计是"每次 send 替换"——如果你连续 send("A") send("B") send("C")，一个恰好在 send("A") 后 send("B") 前被 poll 的 receiver 会直接看到 "C"，"A" 和 "B" 对它不存在。

这是 feature 不是 bug——watch 明确声明"只关心最新值"。但如果你的消费者需要处理每一个变化，watch 不合适——用 broadcast。

选 watch 前问自己：我能接受中间值丢失吗？答案是 Yes → watch；No → broadcast。

watch vs broadcast 的选型对比表

维度	watch	broadcast
存储	1 个当前值（`RwLock<T>`）	ring buffer（N 个值）
消息完整性	跳过中间值	保留所有（除非 Lagged）
内存	恒定（1 × sizeof(T)）	恒定（N × sizeof(T)）
sender	可 clone	可 clone
receiver	多个、各自 seen_version	多个、各自 next position
典型场景	配置 / 状态	事件广播
T 要求	无 `Clone` 约束——`borrow()` 返回的是 `Ref<'_, T>` 借用守卫	`T: Clone`——每个 receiver 都拿一份克隆（`clone_value`），`rem` 减到 0 时槽里的原值直接被丢弃

选 watch 时用错 broadcast 的征兆：消息堆积、receiver 处理不过来、频繁收到 Lagged。 选 broadcast 时用错 watch 的征兆：receiver 发现消息少了（中间被跳过）。

这两种错误都很隐蔽——代码跑得起来、只是语义不对。理解 channel 语义 = 正确使用 channel。

13.8 Cancellation Safety：select! 里 channel 的关键特性

Cancellation safety（取消安全）是 async Rust 里一个令人头疼的概念——它关注"如果一个 Future 被 drop 在半途、会不会丢消息？"。对 channel 来说这个问题特别重要：在 select! { msg = rx.recv() => ... } 里、如果另一个分支先完成、正在 rx.recv() 的 Future 会被 drop——如果此时消息已经被"取出"但还没返回给用户、就会丢。Tokio 的 channel 设计保证 recv().await 是 cancellation-safe 的——不会因为中途被 drop 而丢消息。这个性质对 select! 的正确使用至关重要。

Cancellation safety 是 Rust async 的一个独特概念——一个 Future 被 drop 时不应该丢失信息。

tokio::select! {
    msg = rx.recv() => { /* 处理消息 */ }
    _ = timeout_fut => { /* 超时 */ }
}

如果 timeout 先 Ready，rx.recv() 被 drop。它应该保证没有消息丢失——即使有 sender 在 drop 时刚发出消息，下一次 rx.recv() 也应该收到。

Tokio 各 channel 的接收侧都标了 cancel-safe（下面每条都能在源码的 # Cancel safety 注释里找到，唯独 oneshot 没写这一节）：

mpsc recv：已入队消息一定能被下次 recv 拿到
broadcast recv：drop 不丢 slot
watch changed：版本号机制保证下次 changed() 能检测到
oneshot recv：drop 前值如果已到，存在 UnsafeCell 里——下次 poll 能拿到（但 oneshot 用法里通常不会这么写；注意 tokio/src/sync/oneshot.rs 里没有 # Cancel safety 文档段，这条是从实现推的，不是官方承诺）

这个特性对 select! 和 timeout 是必须的——没有它，"可取消的异步代码"无法正确工作。

自己实现 channel-like Future 时要格外注意 cancellation safety——Tokio 文档对每个 async method 是否 cancel-safe 都有明确标注。第 14 章（select!）会再深入讲。

channel-specific cancellation safety 清单

mpsc::Receiver::recv：✅ cancel-safe
mpsc::Sender::send：⚠️ 半 safe——如果 send 被 cancel 时值还没入队，值消失（你的 value 变量被 drop 了）。这不算"message loss"（message 没发出、也不丢失任何已发的）但数据还是丢了
mpsc::Sender::reserve：值不会丢（它压根不接收值），但官方文档并没有称它 cancel-safe——tokio/src/sync/mpsc/bounded.rs:1072 那段 # Cancel safety 的原话是"makes you lose your place in the queue"：取消会丢掉你在等待队列里的位置
broadcast::Sender::send：不涉及——它是同步方法，不产生可被取消的 Future
broadcast::Receiver::recv：✅ cancel-safe
oneshot::Receiver (作为 Future)：✅ 按实现是 cancel-safe（文档没写 # Cancel safety 段，见上）
watch::Receiver::changed：✅ cancel-safe

清单看起来都 safe——但细节藏在 "send" 类方法里。实际使用中大多数场景是 receiver 侧被 cancel——这几乎总是安全的。sender 侧被 cancel 罕见但可能漏数据。

真实案例：哪类代码会踩 cancellation 坑

// 看起来没问题
let mut rx = open_stream();
loop {
    let msg = tokio::select! {
        m = rx.recv() => m,
        _ = shutdown.recv() => break,
    };
    process(msg).await;  // ← 这里可能很久
}

process(msg).await 不是 cancel-safe 的一部分——如果 shutdown 到达时 process 正在跑，这个 msg 已经从 rx 取出来、但 process 没完成。取决于 process 的逻辑，msg 的处理可能丢失。

注意：上面这段代码里 process(msg).await 本来就在 select! 之外，所以它不会被 shutdown 分支取消——真正的问题不是"被取消"，而是关闭流程不等它跑完：break 之后循环结束，如果整个 runtime 紧接着 shutdown，正在跑的 process 可能被连根拔起。

再看把处理写进分支体里的写法：

// ⚠️ 处理逻辑在 select! 的分支体内
loop {
    tokio::select! {
        Some(msg) = rx.recv() => {
            process(msg).await;      // ← 分支体跑完之前，shutdown 分支不会再被 poll
        }
        _ = shutdown.recv() => break,
    }
}

这里 process(msg).await 是在 select! 内部 await 的。select! 一旦选定分支就会跑完该分支体（不再取消它），所以这段本身不丢消息，代价是处理期间 shutdown 得不到响应；但只要你在分支体里再嵌一层 select! 或把它塞进 timeout(...)，msg 就会连同半途的处理一起蒸发——消息已经离开 channel、却没被处理完。

稳妥的写法：把"取出消息"和"处理消息"分开，取出后立刻脱离任何可能被取消的上下文，并在收到 shutdown 后等在途任务收尾（比如用 JoinSet 或一个计数信号量）。第 14 章会把这条规则展开讲。

cancellation safety 的来源：Future 的 Drop 行为

为什么 Tokio channel 都 cancel-safe？核心是 Drop impl 小心恢复状态。

以 mpsc 的 recv 为例。1.40.0 里 recv 是 async fn，内部就是 poll_fn(|cx| self.chan.recv(cx)).await（tokio/src/sync/mpsc/bounded.rs:240），并没有单独的 Recv 类型和 Drop impl；下面把它写成手写 Future 只是为了说明"drop 时没有东西要回滚"：

// 示意，非源码：mpsc 里不存在这个 Recv 类型
impl<'a, T> Future for Recv<'a, T> {
    type Output = Option<T>;
    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<T>> {
        // 尝试 pop
        if let Some(v) = self.chan.try_pop() {
            return Poll::Ready(Some(v));
        }
        // 注册 Waker
        self.chan.rx_waker.register(cx.waker());
        // 双检查
        if let Some(v) = self.chan.try_pop() { /* ... */ }
        Poll::Pending
    }
}

impl<'a, T> Drop for Recv<'a, T> {
    fn drop(&mut self) {
        // 关键：drop 不改变 channel 状态、不 "消费"任何消息
        // 我们注册的 Waker 会被下次 register 覆盖或自然清理
    }
}

关键观察：drop 时几乎没有要做的事。注册的 Waker 留在那里也没事——下次有人 recv 时会覆盖。没有入队操作需要回滚——因为 recv 是读者侧、只读不写。

send 侧（bounded mpsc） 稍微复杂。send 同样是 async fn：先 self.reserve().await 拿 permit，再 permit.send(value)（tokio/src/sync/mpsc/bounded.rs:800）。取消时的回滚发生在它等待的 batch_semaphore::Acquire 的 Drop 里：

// 简化自 tokio/src/sync/batch_semaphore.rs:686
impl Drop for Acquire<'_> {
    fn drop(&mut self) {
        // 已经完成、不在等待队列里：什么都不用做
        if !self.queued {
            return;
        }
        let mut waiters = self.semaphore.waiters.lock();
        // 把自己从等待队列摘掉
        let node = NonNull::from(&mut self.node);
        unsafe { waiters.queue.remove(node) };
        // 已经分到手的部分 permit 还回去
        let acquired_permits = self.num_permits - self.node.state.load(Acquire);
        if acquired_permits > 0 {
            self.semaphore.add_permits_locked(acquired_permits, waiters);
        }
    }
}
// value 是 send 这个 async fn 状态机里的局部变量，随 Future 一起被 drop

关键：acquired permit 要还——否则"虚假占用"一个 permit。value 被 drop 是数据丢失——但这是 sender 侧 cancel 的预期行为（用户明确 cancel 了 send、值就没了）。

这种 "Drop impl 正确回滚中间状态" 是 cancellation safety 的本质。自己写 Future 时总是想一下 Drop 时需要清理什么——漏掉就是 bug。

一个实际生产案例：用 mpsc 做 task queue

典型 Rust 后端用 mpsc 做 task queue：

// 启动 M 个 worker
let (tx, mut rx) = mpsc::channel::<Task>(1000);
let rx = Arc::new(Mutex::new(rx));

for i in 0..worker_count {
    let rx = rx.clone();
    tokio::spawn(async move {
        while let Some(task) = {
            let mut guard = rx.lock().await;
            guard.recv().await
        } {
            task.execute().await;
        }
    });
}

// producer:
tx.send(task).await?;

但这段代码有问题：mpsc 是 SC（Single Consumer）——用 Mutex 让多 worker 共享是绕过语义，不是推荐做法。

更好的做法：

多生产者、多消费者 → 用 async_channel 或 flume（第三方支持 MPMC）
一个 dispatcher Task 从 mpsc 读 + spawn 到 worker pool—— dispatcher 本身是 SC

// 推荐 pattern
let (tx, mut rx) = mpsc::channel::<Task>(1000);

// dispatcher
tokio::spawn(async move {
    while let Some(task) = rx.recv().await {
        tokio::spawn(async move {
            task.execute().await;
        });
    }
});

第二种更符合 mpsc 语义——每个消息只被 dispatcher 消费一次、处理在另一个 Task。避开了 Mutex<Receiver> 的反模式。

这是生产代码里一个常见的"看起来能跑但设计有问题"案例——识别它需要理解 channel 语义。

13.9 channel 选型：一个决策流程图

这一节是整章的实战落地——当你在真实项目里遇到"这里该用什么 channel"的决策时、本节的流程图能让你 30 秒给出答案。核心决策点就几个：单次还是持续？单消费者还是多消费者？每条都要收到还是要最新值？——三个问题就能定位到 4 种 channel 之一。把这个流程图印下来贴在工位、成为你和团队的共享决策工具。

面对"我应该用哪个 channel"时，按这个流程走：

要发多次消息吗？
├─ 否 → oneshot
└─ 是 → 继续

N 个 receiver 都要收到每条吗？
├─ 是 → broadcast（完整消息）或 watch（最新值）
│   ├─ 消息重要要保留 → broadcast
│   └─ 只要最新状态 → watch
└─ 否 → mpsc

bounded 还是 unbounded？
├─ 需要背压 → bounded mpsc
└─ 不需要背压（接收快，不会堆积）→ unbounded mpsc

大多数场景 mpsc 就够——多 Task 向一个 Task 汇报数据。其余才用其他三种。

具体例子：

Web server 的 request 转发到 worker pool → bounded mpsc
配置热更新给 N 个模块 → watch
日志聚合（多 producer 一个 consumer）→ bounded mpsc（日志不能丢靠的是背压，不是无界堆积；满了要丢还是要等，由发送端显式决定）
事件广播（所有订阅者都要收到事件）→ broadcast
async fn 的返回值 → oneshot

一个常被忽视的选择：mpsc vs broadcast(1) vs watch

三者在"发一个消息通知一个（或多个）接收者"上有重叠：

mpsc::channel(1)：bounded mpsc，容量 1
broadcast::channel(1)：ring buffer 容量 1
watch::channel(initial)：最新值

差别：

mpsc：只能有 1 个 receiver（Receiver 不能 clone）→ 每条消息只被它拿到一次；容量 1 时第二条 send 要等第一条被取走
broadcast(1)：ring buffer 覆盖——如果 receiver 慢、message 被覆盖（Lagged）
watch：总是最新、middle 值被跳过

细微但重要——选错会导致生产 bug。建议针对场景明确语义再选。

select 里写 channel 的推荐模式

// 推荐模式
loop {
    tokio::select! {
        // biased 让 shutdown 优先
        biased;
        _ = &mut shutdown => break,
        
        msg = rx.recv() => {
            let Some(msg) = msg else { break; };  // None = channel 关了
            // 快速消费——复杂处理放 select 外
            // 或 spawn 到另一个 Task 处理、保证 select 循环自身快
            process_fast(msg);
        }
    }
}

几个要点：

biased; —— 让关键分支优先（否则随机选 Ready 分支）
Some / None 区分"消息" vs "channel 关闭"
select! body 内只做快速事——复杂逻辑放外面或 spawn

这种模式是生产代码里 大多数 select 用法的模板——记住它就能覆盖大多数场景。

13.10 bounded vs unbounded mpsc 的实际选择

"bounded（有界）还是 unbounded（无界）"是 mpsc 选型的核心决策。bounded 的好处是提供背压——sender 发送过快时会被挂起、避免消息无限堆积打爆内存；坏处是增加死锁风险（sender 等 receiver、receiver 等某个 resource、resource 等 sender）。unbounded 的好处是绝不阻塞 sender；坏处是没有任何保护——慢 consumer 会导致内存爆炸、服务 OOM。真实生产服务里几乎总是应该用 bounded——"有背压" 是一个服务健康运行的基本要求。

两者 API 几乎一样，但语义差别巨大：

bounded：

channel(N) 创建，容量 N
tx.send(x).await —— 满了 await
有 backpressure：sender 等接收端消费
Semaphore 做 permit 控制

unbounded：

unbounded_channel() 创建
tx.send(x) —— 同步非阻塞，立即返回（接收端已关闭时返回 Err）
无 backpressure：sender 永不等
内部用简单 AtomicUsize 做计数（不是 Semaphore）

bounded 更安全——如果 consumer 卡住、sender 会等、系统自动降速。 unbounded 更危险——consumer 卡住、sender 继续发，内存无限涨直到 OOM。

生产代码建议默认 bounded。用 unbounded 必须有明确理由（性能敏感 + 能保证 consumer 速度）。

一个真实故事：unbounded mpsc 爆内存

2021 年我排查过一次生产 OOM。服务架构：WebSocket 消息 → unbounded mpsc → 处理 worker。

bug：某个连接的 worker 因为 DB 卡住停止消费，websocket 还在疯狂收消息并 push 到 mpsc。mpsc 是 unbounded —— 无上限堆积。几分钟后 OOM。

修复：换成 channel(10000) —— 满了 websocket sender await、自然反压到连接层。如果 10000 条都堆着说明这个 client 太活跃、可以暂停接收。

这是 unbounded 最典型的生产事故模式——consumer 卡死 → sender 无阻塞堆积 → OOM。bounded 用 backpressure 防止这种雪崩。

生产 Rust 代码 review 时，看到 unbounded_channel 要特别警惕——除非有明确理由（比如日志，永不能丢 + consumer 保证快），默认 bounded 更安全。

一个鲜为人知的 API：`mpsc::Sender::try_reserve`

bounded mpsc 提供 reserve 和 try_reserve —— 提前占用一个 permit 但不立刻发消息：

let permit = match tx.try_reserve() {
    Ok(p) => p,
    Err(_) => { 
        // 现在没 permit、先做别的
        continue;
    }
};
// 有 permit 了、接下来可以肯定 send 成功
permit.send(make_value()); // 同步、不等

使用场景：

"计算值很贵、只在 sender 有空间时才做"—— 先 reserve 确认有空间、再计算
原子性的"发或不发"判断——比 send 更灵活

这类 API 在构建高性能 pipeline 时很有用——让你避免"算了才发现 channel 满了、白算"。

13.11 和这个系列的其他专栏的关联

关于 channel 我还想多写几段结构化的思考——这些思考不容易塞进前面具体节的实现讨论里、但值得读者带走。

第一，channel 的"类型化安全"是 Rust 相对其他语言的显著优势。Go 的 chan T 也是类型化的、但 Rust 的 channel 额外利用了所有权系统：发送一个值到 channel 就转移了所有权、接收方收到的是独立的值、发送方失去访问权——这从根本上消除了"数据竞争"这类问题。这是所有权规则在编译期就保证了的性质，不靠运行时检查或编码约定。写过 Java 或 Python 的开发者可能会惊讶于 Rust 这种"消息传递而不是共享状态"的编程模式在真实项目里用得有多顺手——一旦习惯了、几乎不想回到共享可变状态的世界。

第二，channel 容量是服务健康的重要调节器。当你给一个 mpsc 设定 bounded 容量 100、这个数字本身就是一个限流器——sender 到达极限就得等 receiver 消化。这让 channel 成为天然的背压机制、不需要额外写限流代码。这个性质在真实服务里被大量使用：接收请求的任务给处理任务 mpsc 容量 1000 → 处理慢时新请求自动被挡住 → 不会打爆内存。理解这个"容量即背压"的含义能让你在设计服务架构时更有章法。

第三，channel 是调试异步代码的"最佳断点"。当你想理解一个复杂异步应用里数据怎么流动时、在 channel 的 send 和 recv 处加 tracing 日志就能清晰看到"谁给谁发了什么"。把 channel 作为"消息边界"来设计架构、让你的服务有天然的可观测性——日志 / metrics / tracing 都能自然地对齐 channel 事件。这是很多经验丰富的 Tokio 开发者总结出的心得：尽量用 channel 解耦组件、少用 Mutex 共享状态——前者可观测性强、后者黑盒。

第四，channel 的设计哲学和微服务架构高度相关。你读本章实现细节时可能觉得"channel 很抽象"——但如果把每个 task 想象成一个微服务、channel 就是服务之间的消息队列、那这些概念立刻落地：mpsc 是点对点队列、broadcast 是 pub/sub、watch 是配置中心、oneshot 是 request/response。在单进程里用 channel 训练自己的"消息驱动架构"思维、等你做分布式系统时会发现心智模型直接迁移——只是消息跨了进程边界而已。

本章讲的 channel 设计——不同数据结构 + 上一章的 Semaphore / Notify 的组合 —— 和 《Vue 3 设计与实现》第 15 章（状态管理：Pinia 内核） 里讲的 Pinia store 订阅机制是同构思路：Pinia 的 store 也是一个"值 + 订阅者列表 + 变更通知"——watch channel 的前端版。"observer pattern + 最新值" 是跨技术栈的通用设计。

《Rust 编译器与运行时揭秘》第 7 章（Trait 静态分发：零成本抽象的编译器实现）与第 8 章（Trait Object 与虚表） 里讲的 Box<dyn Trait> 和本章 broadcast 的 Box<[RwLock<Slot<T>>]>都是"编译期确定布局 + 运行时动态内容"的典型。这种 "Box 固定大小数组" 是 Rust 做"运行时决定大小但此后不变"的标准做法。

13.12 实战案例：用 mpsc + oneshot 桥接同步世界

真实项目里我们经常需要把同步的东西（SQLite 驱动、老的 C 库 FFI、CPU-heavy 的处理）和异步的业务代码协作起来。直接在 async fn 里调同步函数会阻塞整个 worker——常见解法有两条：spawn_blocking 包住每次调用，或者开一条专用线程 + channel 桥接。下面展示后者的完整模板（并在小节末尾说明它为什么有时优于前者）。

一个常见生产模式：把同步的 blocking 库桥接成 async 接口。mpsc 是核心：

// 假设 some_sync_lib::DbClient 是同步阻塞的、不能直接在 async 里用
// Rows / Result 换成你自己库的类型

use tokio::sync::{mpsc, oneshot};

pub struct AsyncDb {
    tx: mpsc::Sender<Command>,
}

enum Command {
    Query(String, oneshot::Sender<Result<Rows>>),
    Shutdown,
}

impl AsyncDb {
    pub fn new() -> Self {
        let (tx, mut rx) = mpsc::channel(100);
        
        // 专用线程跑同步 DB 客户端
        std::thread::spawn(move || {
            let mut db = some_sync_lib::DbClient::connect("...");
            while let Some(cmd) = rx.blocking_recv() {
                match cmd {
                    Command::Query(sql, reply) => {
                        let result = db.query(&sql);
                        let _ = reply.send(result);
                    }
                    Command::Shutdown => break,
                }
            }
        });
        
        Self { tx }
    }
    
    pub async fn query(&self, sql: String) -> Result<Rows> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx.send(Command::Query(sql, reply_tx)).await?;
        reply_rx.await?
    }
}

关键元素：

mpsc channel 作为命令队列（async → sync 方向）
oneshot 作为返回值（sync → async 方向）
一个 std thread 跑同步库
blocking_recv 在 std thread 里阻塞等命令

这个模式覆盖大量"async 代码要用老同步库"场景：Redis 同步客户端、老版 DB 驱动、FFI 调用。两种 channel 的组合轻松解决。

一个深入问题：为什么不用 tokio::task::spawn_blocking 包裹每次查询

// 不用 mpsc，每次查询 spawn_blocking
pub async fn query(&self, sql: String) -> Result<Rows> {
    let db = self.db.clone();
    tokio::task::spawn_blocking(move || {
        db.query(&sql)
    }).await?
}

两个问题：

DbClient 不能跨线程安全共享（某些同步库的连接是 !Send 的）—— spawn_blocking 要求 Send
每次查询都可能落在 blocking 池里不同的线程上——self.db.clone() 要求 DbClient 能廉价 clone、还能被多条线程同时使用；不少同步客户端做不到，或者 clone 一次就意味着新建一条连接

mpsc + 专用线程的好处：DbClient 在专用线程上始终活着，连接不断、状态保留。这是"长生命周期资源 + 异步访问"的经典解。

13.13 为什么 Tokio 的 mpsc 不叫 "unbounded"

这是一个API 设计上的精心选择——Tokio 把 bounded mpsc 作为默认 tokio::sync::mpsc、unbounded 版本放在 tokio::sync::mpsc::unbounded_channel、专门要求用户写出 unbounded 这个单词才能用。这种"让安全版本更容易访问、有风险的版本需要明确写出来"的 API 设计思路在 Rust 生态里反复出现（unsafe 块、Box::leak、get_unchecked）——让用户默认走对的路径、想要越界操作时让他明确确认。

细心读者会问：Tokio 的 mpsc 有 bounded 和 unbounded 两个版本，但两者 API 几乎一样、返回类型都叫 Sender / Receiver。只是创建函数不同（channel(N) vs unbounded_channel()）。

实际上它们不是同一种类型：

mpsc::Sender<T> / mpsc::Receiver<T>：bounded 版
mpsc::UnboundedSender<T> / mpsc::UnboundedReceiver<T>：unbounded 版

类型分离的原因：bounded 的 send 是 async（要等 permit），unbounded 的 send 是同步（永不等）。两种不同的 API 必须不同类型。

这给库作者一个启示：类型系统是 API 契约的载体——把语义差异编码进类型、编译器帮你抓住误用。

13.14 一个深层的观察：channel 是分布式系统的最小单元

这一节把本章的视野从单机推到分布式——你会看到"channel"这个概念在分布式系统里以另一种形式出现。单机 channel 是"两个 task 之间的消息管道"——分布式消息队列（Kafka、RabbitMQ、NATS）是"两个服务之间的消息管道"。语义惊人地一致：都有 producer / consumer / 顺序保证 / 背压 / 订阅模式。一个 Kafka topic 其实就是一个 "持久化 + 分布式 + 多消费者" 版的 broadcast channel——这个映射让你能用一套心智模型覆盖单机和分布式。

读完本章你应该意识到：channel 是"同步 + 通信"的最小打包——两个角色通过它交换信息。推而广之，分布式系统的消息队列（Kafka、RabbitMQ）、微服务的 RPC、甚至 HTTP/gRPC——都是 channel 的分布式版本。

Kafka 对应 broadcast：多消费者各自 offset、不丢历史
Redis pub/sub 对应 broadcast：发即广播、不存消息，订阅之前发的收不到
gRPC unary call 对应 oneshot：request + response 一次性
gRPC bidi streaming 对应 mpsc × 2：双向 mpsc

这种 "语义同构"让你能用一套心智模型覆盖单机 + 分布式。学会 Tokio channel，你就学会了分布式消息 system 的底层——细节不同但骨架一样。

本专栏专注单机 Tokio，但这种推广思维是你从本章带走的最大财富之一。

13.15 本章小结

channel 是 Tokio 异步编程里你每天都会用的核心工具——学完本章你对四种 channel 的语义、实现、适用场景、性能特征都有完整认识。关键 take-home：

1. 选 channel 的三个灵魂拷问：（1）单次还是持续？ 单次选 oneshot；（2）单消费者还是多消费者？ 单消费者选 mpsc、多消费者选 broadcast；（3）每条都要还是最新值？ 最新值选 watch。这三个问题在脑子里过一遍、正确答案立刻出来。

2. 成本看结构（记结构，别记数字）：oneshot 最轻（没有队列、没有锁）；mpsc 这一支 unbounded < bounded（后者多一层信号量）< broadcast（尾锁串行化 + 每槽读写锁 + 每个订阅者一份克隆）；watch 不在这条线上——它没有队列，但每次 send 要拿一次写锁、再对 8 个 Notify 逐个 notify_waiters（每个都要拿它自己的内部锁），和 unbounded mpsc 谁轻谁重取决于负载。具体差多少必须在自己的负载上测。

3. 永远用 bounded mpsc（除非你非常确定）：unbounded mpsc 在内存管理上有隐性风险——慢 consumer + 快 producer = 内存无限增长。bounded mpsc 提供背压、一个关键的系统健康屏障。写出带内存保护的代码比写出"看起来更快的"代码重要得多。

4. cancellation safety 是所有 channel 的基本契约——Tokio 各 channel 的接收操作都是 cancel-safe 的（oneshot 文档未写明、按实现成立）、你在 select! / timeout 里用它们不会丢消息。自己写 channel-like API 时也必须达到这个标准、否则会成为难以调试的 bug 源。

下一章（第 14 章）我们讲 select! ——Tokio 最强大也最常被误用的宏。你会看到 select! 怎么协调多个异步分支、为什么 cancellation safety 在这里特别重要、以及哪些模式是推荐的哪些是陷阱。channel 和 select! 的组合是 Tokio 异步编程的"主力武器"——学完这两章你的 async Rust 编程能力会进入下一个台阶。

四种 channel 的成本排序（结构推导，不是实测）

本机没有可复现的基准环境，所以这里不给纳秒数——但仅从前面拆过的实现就能推出大部分先后关系：

channel	一次 send + recv 要做的事
oneshot	写一次 `UnsafeCell` + 一次 CAS 置位 + 一次 `load`，没有队列、没有锁
watch	写锁改值 + 版本号自增 + `BigNotify` 通知（对 8 个 `Notify` 逐个 `notify_waiters`，每个都要拿内部锁，`tokio/src/sync/watch.rs:354-358`、`tokio/src/sync/notify.rs:663`）；读端拿读锁借用，不复制值
unbounded mpsc	`fetch_add` 抢槽 + 写块内槽位 + `AtomicWaker` 唤醒
bounded mpsc	以上再加一次 batch_semaphore 的 acquire / release
broadcast	`Mutex<Tail>` 串行化 + 槽位 `RwLock` 写锁 + 唤醒全部等待者；读端读锁 + 克隆值 + `rem` 计数

能从结构推出的排序：oneshot 最轻；unbounded mpsc < bounded mpsc < broadcast。watch 单列：它没有队列、读端不复制值，但写端每次都是 1 次写锁 + 8 次 Notify 内部加锁，和 unbounded mpsc 相比谁更轻，结构上推不出定论。

这些结构差异跨机器稳定；具体差多少则完全取决于你的消息类型、并发度和竞争程度——要数字就在自己的负载上测。

读完本章你应该能回答：

mpsc 为什么用链表不用 ring buffer？ 答：要支持 unbounded + 单消费者无竞争。
broadcast 为什么用 ring buffer？ 答：多消费者并发读 + 固定内存 + 允许 slow 消费者 lag。
oneshot 为什么只用一个 AtomicUsize？ 答：一次性语义简单，RX_TASK_SET / VALUE_SENT / CLOSED / TX_TASK_SET 四个 bit 就装得下全部状态。
watch 为什么不保留历史？ 答：订阅语义是"当前值"、保留历史会违背定位、开销也增加。
为什么各 channel 的接收侧都 cancel-safe？ 答：因为 select! 和 timeout 依赖这个、这是生态基础契约。

如果你都能答出——channel 家族你已经掌握。

带走三件事：

四种 channel 的底层数据结构各不同：mpsc = 无锁分块链表（64 位平台上每块 32 个连续槽位）、oneshot = 单 atomic 状态字 + 一个值槽、broadcast = 环形缓冲 + 每槽一把 RwLock + rem 引用计数、watch = RwLock<T> + 版本号 atomic。每种选择都针对自己的语义最优
前面章节的 batch_semaphore / Notify / AtomicWaker 在 channel 里被有选择地复用——bounded mpsc 用 batch_semaphore 做背压、用 AtomicWaker 唤醒 receiver；watch 用 BigNotify 唤醒订阅者；broadcast 和 oneshot 则完全自建（broadcast 的等待链表挂在 Tail 里，oneshot 直接存两个裸 waker 槽）。除 broadcast 靠 Mutex<Tail> 串行化之外，其余几种都把关键状态压进 AtomicUsize 做协调
Tokio 各 channel 的接收侧都是 cancellation-safe 的（mpsc::Sender::send 这类发送侧不是：被取消时值会被丢掉）——select! 和 timeout 能正确工作的前提。自己写 channel-like API 时必须达到这个标准

用 7 个签名模式看 channel

回顾第 8 章总结的 Tokio 7 大签名模式，看 channel 如何体现：

分离独占和共享：Chan 里 tx（多写者、原子）和 rx_fields（单读者、UnsafeCell）按访问模式分开，tx 与 rx_waker 还各自 CachePadded
单 AtomicUsize 装多字段：oneshot 的四个状态位、watch 的"版本号 + 最低位 closed"
侵入式链表：broadcast 的接收者等待链表（挂在 Tail 里）、Semaphore 的 waiters
vtable + 指针作标识：channel 这里基本没用到——它们都是单态泛型
fast path 的借用语义版本：AtomicWaker::register_by_ref、watch 的 borrow() 返回 Ref<'_, T> 而不是克隆
CAS loop 封装成 helper：oneshot 的 State::set_complete（set_closed 等其余 setter 用的是 fetch_or）
分批 wake 避开锁内副作用：batch_semaphore 的 WakeList——先 drop(waiters) 再 wake_all()

7 个模式里 6 个在 channel 出现——channel 是 Tokio 内部设计原则的完整展示。读懂 channel，你会看到 Tokio 内部那套"设计语言"的主要词汇。

第三方 channel 的选项

Tokio 自带的 4 种 channel 覆盖绝大多数场景。但有几个第三方 channel crate 在特定场景值得考虑：

flume：

兼容 sync 和 async（同一个 channel 两种 API）
比 Tokio mpsc 在纯 sync 场景更快
适合"部分代码 sync、部分 async 共用 channel"

async-channel：

和 Tokio mpsc 独立、不依赖 runtime
smol 运行时生态首选
API 轻微差异但语义相近

crossbeam-channel：

完全 sync，不是 async
高性能多生产者多消费者
可以和 Tokio 混用（通过 spawn_blocking 桥接），但不 async-native

tokio_util::sync::PollSender（tokio-util crate）：

把 tokio::sync::mpsc::Sender 包装成 Sink（futures crate 的 trait）
用于需要 Sink 抽象的库

选择标准：优先 Tokio 原生、除非有具体理由换。Tokio mpsc 对绝大多数场景已经足够快 + 和生态兼容。

一个总结：Tokio 的 channel 如何对齐 Go

维度	Go	Tokio
mpsc	`chan T`	`tokio::sync::mpsc`
广播（每个消费者都收到）	-	`broadcast`（Go 的 chan 虽可多消费者，但一条消息只给一个，广播要手动）
最新值	-	`watch`（Go 里通常用 atomic + channel 组合）
一次性	-	`oneshot`（Go 里用 `chan struct{}` 加 select）
select	`select { ... }`	`tokio::select! { ... }`（下一章）

Tokio 的 channel 家族比 Go 更丰富——Go 是"一个 chan 解决所有"、Tokio 是"针对场景提供专门优化"。各有优劣：Go 心智负担低但可能写不出最优代码、Tokio 需要选对但选对后代码更清晰高效。

一个深度思考：为什么 Rust 生态没有标准 "channel trait"

Go 的 chan T 是语言关键字——所有人用同一套语义。Rust 生态里 Tokio mpsc、flume、crossbeam、async-channel 的 API 各不相同——没有统一 trait。

为什么：

各 channel 有独特 API——mpsc 有 reserve、broadcast 有 resubscribe、watch 有 borrow——很难抽象成统一 trait 不丢语义
async 和 sync 的 channel 签名天然不同——sync recv 直接返回结果（如 std::sync::mpsc 的 Result<T, RecvError>）、async recv 返回 impl Future<Output=Option<T>>
有过尝试但没统一起来：futures crate 里的 Stream（接收侧）/ Sink（发送侧）算最接近的，但 Tokio 的 channel 并不直接实现它们（要靠 tokio-stream 的 ReceiverStream 或上面的 PollSender 包一层）
统一 trait 会限制优化——每 channel 根据语义特化实现、trait 抽象会丢失

Rust 生态选择了"具体类型优于 trait 抽象"。代价是学习曲线（要学每个 channel 的具体 API）、好处是每个都能被深度优化。

这是 Rust 哲学的一个反复主题——"多个具体实现" + "零成本抽象" vs "统一 trait + 运行时分派"。前者需要更多工作但性能更高。Tokio channel 是前者的代表。

下一章我们进入 select! 宏——看它如何展开成一段确定的 state machine、如何选择 Ready 分支、biased; 关键字改变了什么。select! 是 channel + timer + cancellation 的集大成。

延伸阅读

Tokio 源码：tokio/src/sync/mpsc/chan.rs
Tokio 源码：tokio/src/sync/broadcast.rs
Tokio 源码：tokio/src/sync/oneshot.rs
Tokio 源码：tokio/src/sync/watch.rs
《Vue 3 设计与实现》第 15 章：Pinia store 订阅机制和 watch channel 的同构
《Rust 编译器与运行时揭秘》第 7 章：Box dyn Trait 和固定大小数组的内存布局