Tokio 源码深度解析

第12章异步 Mutex / RwLock / Semaphore

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

本章要点

tokio::sync::Mutex<T> 的真实定义：3 个字段——Semaphore (内部 permits=1) + UnsafeCell<T> + 可选的 tracing span。Mutex 就是"一个只能有一个持有者的 Semaphore"
Semaphore 是 Tokio 所有同步原语的 building block——Mutex、RwLock、OnceCell、有界通道的背压都在其上构建（Notify 是独立实现）
底层 batch_semaphore::Semaphore 的结构：waiters: Mutex<Waitlist>（侵入式链表 + 同步 Mutex 保护）+ permits: AtomicUsize（可用 permit 数）
Tokio Mutex 是公平锁（FIFO 排队）——和 std::sync::Mutex 的非公平语义有区别
MutexGuard::drop 调 lock.s.release(1)——释放 permit、触发下一个等待者
RwLock = Semaphore(N) + 读写各占 permit——写锁占 N permit、读锁占 1 permit
尽量不要跨 .await 持锁——不只是因为"性能差"，还因为可能死锁：被等待的 Future 想要获取同一把锁

12.1 一个直觉问题：async 需要自己的 Mutex 吗

这个看似简单的问题其实触及异步编程的根本哲学：异步并发 vs 同步并发是两种不同模型、不能简单互换工具。同步 Mutex 的 lock 操作阻塞线程——在异步世界里这相当于"阻塞 worker 线程上所有其他 task"——代价巨大。所以异步世界需要"让出到调度器"的 lock 语义——这就是 Tokio Mutex 存在的理由。

在深入具体实现前，想一下：std::sync::Mutex 已经存在、为什么 Tokio 还要搞一套？

原因在"锁被占时会发生什么"：

std::sync::Mutex：拿锁失败 → 当前OS 线程进 kernel wait 队列、CPU 让给其他线程。整个线程被挂起。

Tokio 场景：一个 worker 线程可能跑着几千个 Task。如果一个 Task 被 std Mutex 挂起 → 整个 worker 挂起 → 上面几千个 Task 全被卡住。灾难。

tokio::sync::Mutex：拿锁失败 → 当前 Task 返回 Pending、注册 Waker、让出给 runtime。OS 线程不被挂起——它继续跑别的 Task。锁好了后 Waker 唤醒那个 Task。

这是 async 同步原语的本质：把"阻塞线程"替换成"挂起 Task"。Rust std 没有 Task 概念——所以必须 Tokio 自己提供 async 版本。

这就是为什么 tokio::sync::* 存在。不是重复造轮子——是必要的。读本章前想清楚这点——后面所有设计都围绕"Task 让步 + Waker 唤醒"展开。

12.2 Tokio 同步原语全景

Tokio 提供的同步原语是系统级工具箱——每个原语解决一类特定的"多 task 之间如何协作"问题。Mutex 解决互斥、RwLock 解决读写分离、Semaphore 解决限流、Notify 解决条件唤醒、Barrier 解决同步点、oneshot/mpsc/broadcast 解决消息传递。学 async 编程一定要对这张全景图有清晰认知——遇到问题时才能选对工具。本节把这张图画清楚——后面每一节再深入讲具体原语。

Tokio sync 模块提供了一整套异步同步原语：

原语	用途	底层实现
`Mutex<T>`	互斥访问	Semaphore(1) + UnsafeCell
`RwLock<T>`	读写锁	Semaphore(N) + 读写 permit 数分配
`Semaphore`	信号量（限流、并发控制）	直接就是 batch_semaphore
`Notify`	条件变量（wake up 一个 / 多个等待者）	独立实现，但思路相似
`Barrier`	集合屏障	同步 Mutex（`crate::loom::sync::Mutex`）保护计数 + `watch` 通道广播代次
`OnceCell<T>`	一次初始化	基于 Semaphore

核心洞察：Semaphore 是锁类原语的基础。理解了 Semaphore，就理解了 Mutex、RwLock 等。所以本章先讲 Mutex（最小案例），再讲 Semaphore（基础机制），再讲 RwLock / Notify（变种）。

12.3 Mutex 的惊人简单

Tokio 的 Mutex 实现之简单程度让人意外——它就是 Semaphore(1) 加一个 UnsafeCell。"互斥锁 = 容量为 1 的信号量"是并发理论里的基础结论——信号量这个原语正是 Dijkstra 在 1960 年代提出的，二元信号量（binary semaphore）与互斥量的对应关系是它最早的应用之一。Tokio 把这个结论工程化成了一个两字段的 struct。这种"用更基础的原语构造上层抽象"是 Tokio 整个 sync 模块的核心设计哲学——Semaphore 是最底层、Mutex/RwLock/OnceCell 都建立在它之上。

打开 tokio/src/sync/mutex.rs。原样：

// 来源：tokio-rs/tokio · tokio/src/sync/mutex.rs:133 (tokio-1.40.0)
pub struct Mutex<T: ?Sized> {
    #[cfg(all(tokio_unstable, feature = "tracing"))]
    resource_span: tracing::Span,
    s: semaphore::Semaphore,
    c: UnsafeCell<T>,
}

3 个字段（非 tracing 下 2 个有效字段）：

s: semaphore::Semaphore —— 一个 permit = 1 的 Semaphore
c: UnsafeCell<T> —— 存放被保护的数据
resource_span（可选） —— tracing 观测

这就是 Tokio Mutex 的全部结构。让我们想想它的语义：

初始化：Semaphore::new(1) —— 一个可用 permit
lock().await → 调 semaphore 的 acquire() → 如果拿到 permit，返回 MutexGuard；否则等
MutexGuard drop → semaphore.release(1) —— 归还 permit

Mutex = "只有一个 permit 的 Semaphore 包装可变数据"。这是极简、正确、通用的设计。

`new` 的实现

// 示意，非逐字：tokio/src/sync/mutex.rs:338
pub fn new(t: T) -> Self where T: Sized {
    let s = semaphore::Semaphore::new(1);
    Self {
        c: UnsafeCell::new(t),
        s,
        // ...
    }
}

几行代码。构造一个 Semaphore(1)、包装数据。

`lock` 的实现

// 示意，非逐字：tokio/src/sync/mutex.rs:434
pub async fn lock(&self) -> MutexGuard<'_, T> {
    let acquire_fut = async {
        self.acquire().await;
        MutexGuard {
            lock: self,
            // ...
        }
    };
    // ...
    acquire_fut.await
}

async fn——说明 lock 是 yield 点。内部 await self.acquire()，而它调的是 self.s.acquire(1)（tokio/src/sync/mutex.rs:654）。注意这里的 s 直接就是 batch_semaphore::Semaphore——tokio/src/sync/mutex.rs:3 写着 use crate::sync::batch_semaphore as semaphore;，Mutex 根本不经过公开的 tokio::sync::Semaphore 那层带 ll_sem 字段的门面。一个 permit 拿不到就挂起。

拿到 permit 后构造 MutexGuard 返回。MutexGuard 是一个 RAII guard——drop 时自动释放。

`try_lock` 的非阻塞尝试

// 示意，非逐字：tokio/src/sync/mutex.rs:678
pub fn try_lock(&self) -> Result<MutexGuard<'_, T>, TryLockError> {
    match self.s.try_acquire(1) {
        Ok(()) => {
            let guard = MutexGuard { lock: self, /* ... */ };
            Ok(guard)
        }
        Err(_) => Err(TryLockError(())),
    }
}

非 async——try_acquire 是同步方法，要么成功要么失败返回 Err。不等、不让步、立刻返回。用在"我试试拿锁，拿不到就做别的"场景。

`MutexGuard` 的 Drop

// 来源：tokio/src/sync/mutex.rs:963（节选，略去 tracing 部分）
impl<T: ?Sized> Drop for MutexGuard<'_, T> {
    fn drop(&mut self) {
        self.lock.s.release(1);
    }
}

一行——把 permit 还回 Semaphore。这触发等待队列里最前面的 Task——第 12.4 节会看到 release 如何唤醒等待者。

为什么设计这么简单

核心洞察：并发控制的本质是 "permit"——你有 permit 就能做事、没有就等。Mutex 只是"permit 数 = 1"的特例、RwLock 是"读 permit + 写 permit"的扩展、限流器是"permit 数 = N"的应用。

用一个通用 primitive 实现所有同步原语——减少重复代码、减少 bug 可能性、让每个原语都继承基础的正确性。这是软件设计最珍贵的美学。

`Send` 和 `Sync` 的实现

Mutex 需要手动实现 Send / Sync（tokio/src/sync/mutex.rs:258-tokio/src/sync/mutex.rs:259）：

// 来源：tokio/src/sync/mutex.rs:258
unsafe impl<T> Send for Mutex<T> where T: ?Sized + Send {}
unsafe impl<T> Sync for Mutex<T> where T: ?Sized + Send {}

注意 Sync 那条只要求 T: Send、不要求 T: Sync——因为 Mutex 已经把并发访问串行化了，同一时刻只有一个持有者能碰到 T，所以只需要 T 能跨线程搬运。（对照同处的 unsafe impl<T> Sync for MutexGuard<'_, T> where T: ?Sized + Send + Sync {}，tokio/src/sync/mutex.rs:260——guard 允许多处同时拿 &T，那里才需要 T: Sync。）

为什么需要 unsafe impl：因为 UnsafeCell<T> 本身不是 Sync——必须手动承诺。Tokio 通过互斥保证（同一时刻只有一个 Task 能访问 T）让这个承诺成立。

T: Send 是要求——因为 T 可能被不同 worker 线程访问（Task 被迁移）。Sync 也同理——多个引用可能同时存在（&Mutex<T> 可以 clone 给多处）。

这些 Send/Sync bound 影响你能用什么类型：

Mutex<String>：Send + Sync ✓
Mutex<Rc<T>>：Rc 不是 Send——Mutex<Rc<T>> 也不是 Send——不能 spawn
Mutex<*const T>：裸指针不是 Send——同上

如果你需要保护 !Send 数据，换成 std::sync::Mutex 还是 tokio::sync::Mutex 都不能让它变成 Send——需要 tokio::task::LocalSet 里的 !Send 上下文（用 spawn_local，第 7 章讲过）。

Mutex 的 `UnsafeCell<T>`：为什么需要

Mutex 的第三个字段 c: UnsafeCell<T>。为什么是 UnsafeCell 而不是 T 直接？

因为 Rust 的借用规则不允许 &Mutex<T> 转成 &mut T——& 是共享引用、Rust 默认禁止通过共享引用修改数据。

UnsafeCell 是 Rust 里"内部可变性"的唯一基础原语——它是编译器承认的"即使 &Mutex、内部的 T 也可以通过 unsafe 被可变访问"。Mutex 通过 Semaphore 保证互斥、从而让这个 unsafe 访问在语义上安全。

如果不用 UnsafeCell：Mutex 根本没法工作——多个线程拿着 &Mutex<T>（都是共享引用），没法修改 T。UnsafeCell 是整个 Rust 内部可变性生态（Cell、RefCell、Mutex、RwLock、AtomicXXX）的共同基础。

Mutex 只是 2 个有效字段

再回看 Mutex 的定义（非 tracing 模式）：

pub struct Mutex<T: ?Sized> {
    s: semaphore::Semaphore,
    c: UnsafeCell<T>,
}

就 2 个字段——一个 Semaphore、一个 UnsafeCell<T>。mutex.rs 整个文件上千行，但绝大部分是文档注释和 try_lock_owned / blocking_lock 之类的变体；lock / try_lock / Drop 这三段核心逻辑加起来只有十几行有效代码——这就是"底层原语选对了、上层代码就少"的证明。

对比 Linux kernel 的 mutex 实现（乐观自旋、锁交接等机制一应俱全，代码量大得多）——当然 kernel 有更极端的性能和正确性要求。但对用户态 async 场景，Tokio 的极简设计已经足够且更好维护。

12.4 Semaphore：基础 building block

Semaphore 是 Tokio sync 模块的基石——几乎所有上层原语都建立在它之上。它本身是一个"令牌池"——有 N 个令牌、每次 acquire 消费一个、每次 release 归还一个、令牌用光时新的 acquire 挂起等待。这个简单模型支撑起了令人惊讶多的场景：Mutex（令牌数=1）、RwLock（共享令牌 + 独占令牌）、并发限流器（令牌数=并发上限）、连接池（令牌数=max_connections）。理解了 Semaphore 的实现，你就理解了 Tokio 同步原语的大半。

tokio::sync::Semaphore 的外层（tokio/src/sync/semaphore.rs:392）：

// 来源：tokio/src/sync/semaphore.rs:392
pub struct Semaphore {
    ll_sem: ll::Semaphore,
    // ...
}

就一个 ll_sem: ll::Semaphore——ll 是 batch_semaphore 的别名。外层 Semaphore 是 API 门面、实际逻辑在 batch_semaphore。

batch_semaphore 的 struct（tokio/src/sync/batch_semaphore.rs:35）：

// 来源：tokio/src/sync/batch_semaphore.rs:35
pub(crate) struct Semaphore {
    waiters: Mutex<Waitlist>,
    permits: AtomicUsize,
    // ...
}

2 个核心字段：

waiters: Mutex<Waitlist>（tokio/src/sync/batch_semaphore.rs:36）—— 等待队列（侵入式链表 + 一个 closed 标志）。这里的 Mutex 是 crate::loom::sync::Mutex（tokio/src/sync/batch_semaphore.rs:20）——没开 parking_lot feature 时是 std::sync::Mutex 的一层薄包装，tokio/src/loom/std/mutex.rs:6 的 pub(crate) struct Mutex<T: ?Sized>(sync::Mutex<T>) 只是把 poisoning 那套 API 抹掉；开了 parking_lot feature（full 就包含它，tokio/Cargo.toml:35）则换成 parking_lot 的 Mutex（tokio/src/loom/std/mod.rs:59-70），loom 测试时换成 loom 的版本。无论哪种都是同步锁，绝不是 tokio::sync::Mutex——否则循环依赖
permits: AtomicUsize（tokio/src/sync/batch_semaphore.rs:38）—— 当前可用 permit 数，最低位被借去当 CLOSED 标志（const CLOSED: usize = 1; 在 tokio/src/sync/batch_semaphore.rs:131、const PERMIT_SHIFT: usize = 1; 在 tokio/src/sync/batch_semaphore.rs:135），所以真实 permit 数是这个值右移一位

两者协作完成整个 Semaphore 的语义。

`Waiter` 结构（`tokio/src/sync/batch_semaphore.rs:79`）

// 来源：tokio/src/sync/batch_semaphore.rs:79（节选）
struct Waiter {
    state: AtomicUsize,
    /// :88-90 的 SAFETY：只有在持有等待队列锁时才能碰这个字段
    waker: UnsafeCell<Option<Waker>>,
    pointers: linked_list::Pointers<Waiter>,
    // ...
    _p: PhantomPinned,
}

每个等待 Task 都有一个 Waiter 节点：

state: AtomicUsize —— 还差多少 permit（assign_permits 把它减到 0 就说明凑齐了）
waker: UnsafeCell<Option<Waker>> —— 该 Task 的 Waker（好了要 wake 它）
pointers —— 侵入式链表节点（放进 Waitlist）
PhantomPinned —— 让 Waiter 变成 !Unpin：节点地址挂在链表里，入队后不能被 move

`poll_acquire` 的逻辑

poll_acquire（tokio/src/sync/batch_semaphore.rs:397）是 Semaphore 上的方法，由 Acquire 这个 Future 的 poll 调用。简化逻辑：

// 示意，非逐字：简化自 tokio/src/sync/batch_semaphore.rs:397（略去 tracing 和错误分支）
fn poll_acquire(
    &self, cx: &mut Context<'_>, num_permits: usize,
    node: Pin<&mut Waiter>, queued: bool,
) -> Poll<Result<(), AcquireError>> {
    let mut acquired = 0;
    // 已在队列里的 waiter 只要补齐差额（记在 node.state 上）
    let needed = (if queued { node.state.load(Acquire) } else { num_permits }) << Self::PERMIT_SHIFT;

    let mut lock = None;
    let mut curr = self.permits.load(Acquire);
    let mut waiters = loop {
        if curr & Self::CLOSED > 0 { return Poll::Ready(Err(AcquireError::closed())); }

        let mut remaining = 0;
        let total = curr.checked_add(acquired).expect("...");
        let (next, acq) = if total >= needed {
            (curr - (needed - acquired), needed >> Self::PERMIT_SHIFT)
        } else {
            remaining = (needed - acquired) - curr;   // 只够拿一部分
            (0, curr >> Self::PERMIT_SHIFT)
        };

        // 注意：先拿队列锁，再做那个把 permits 清零的 CAS
        if remaining > 0 && lock.is_none() {
            lock = Some(self.waiters.lock());
        }

        match self.permits.compare_exchange(curr, next, AcqRel, Acquire) {
            Ok(_) => {
                acquired += acq;
                if remaining == 0 {
                    if !queued { return Poll::Ready(Ok(())); }
                    else if lock.is_none() { break self.waiters.lock(); }
                }
                break lock.expect("lock must be acquired before waiting");
            }
            Err(actual) => curr = actual,
        }
    };

    // 已经拿到的那部分先记到 node 上；够了就直接 Ready
    if node.assign_permits(&mut acquired) { /* ... */ return Poll::Ready(Ok(())); }

    // 否则注册 waker、把 node 挂进等待队列头部
    node.waker.with_mut(|waker| {  });
    if !queued { waiters.queue.push_front(node_ptr); }
    Poll::Pending
}

几个容易读漏的细节：

permits 里塞了一个标志位：const CLOSED: usize = 1;（tokio/src/sync/batch_semaphore.rs:131）、const PERMIT_SHIFT: usize = 1;（tokio/src/sync/batch_semaphore.rs:135）——最低位当"信号量已关闭"用，真正的 permit 数存在高位。所以代码里到处是 << PERMIT_SHIFT / >> PERMIT_SHIFT
它能"部分拿"：batch_semaphore 的 batch 指一次可以等多个 permit（源码注释："permits waiting on multiple permits at once"）。permits 不够时它会把现有的全部拿走（next = 0），把 remaining 记在 waiter 上继续等——而不是一个都不拿。这样一次要很多 permit 的大请求不会被源源不断的小请求饿死
加锁顺序是刻意的：源码注释写着必须先拿队列锁、再做那个 CAS——否则"先扣 permits 再拿锁"的窗口里别人 release 的 permit 会被漏掉
Fast path 无锁：remaining == 0 且不在队列里时，一次 CAS 成功就直接 Poll::Ready(Ok(()))，根本不碰 waiters 那把锁

只有真的需要等才走 Mutex——高并发但低竞争时，锁基本不会被摸到。

`release` 的逻辑

release（tokio/src/sync/batch_semaphore.rs:231）本身只有三行——它做的第一件事是拿锁，而不是 fetch_add：

// 来源：tokio/src/sync/batch_semaphore.rs:231
pub(crate) fn release(&self, added: usize) {
    if added == 0 {
        return;
    }

    // Assign permits to the wait queue
    self.add_permits_locked(added, self.waiters.lock());
}

真正的逻辑在 add_permits_locked（tokio/src/sync/batch_semaphore.rs:306，下面是简化）：

// 示意，非逐字：简化自 tokio/src/sync/batch_semaphore.rs:306
fn add_permits_locked(&self, mut rem: usize, waiters: MutexGuard<'_, Waitlist>) {
    let mut wakers = WakeList::new();
    let mut lock = Some(waiters);
    let mut is_empty = false;
    while rem > 0 {
        let mut waiters = lock.take().unwrap_or_else(|| self.waiters.lock());
        'inner: while wakers.can_push() {
            match waiters.queue.last() {
                // assign_permits 直接把 permit 记到 waiter 身上，rem 相应减少
                Some(waiter) => if !waiter.assign_permits(&mut rem) { break 'inner; },
                None => { is_empty = true; break 'inner; }
            };
            let mut waiter = waiters.queue.pop_back().unwrap();
            if let Some(waker) = /* 取走 waiter 的 waker */ { wakers.push(waker); }
        }

        // 队列空了还有剩，才真的加回 self.permits
        if rem > 0 && is_empty {
            self.permits.fetch_add(rem << Self::PERMIT_SHIFT, Release);
            rem = 0;
        }

        drop(waiters);       // 先放锁
        wakers.wake_all();   // 再 wake
    }
}

三个关键点：

permit 优先"直接交接"给等待者，而不是先还回 self.permits 再让等待者去抢——assign_permits 把 permit 记在 waiter 的 state 上。只有队列被喂饱之后剩下的才 fetch_add 回计数器。这是公平性的机械保证：排队的人永远比新来的先拿到
queue.push_front 入队、queue.last() / pop_back() 出队——一头进另一头出，就是 FIFO
WakeList 分批、先放锁再 wake：wakers.can_push() 满了就退出内层循环，drop(waiters) 之后才 wake_all()。release 路径不在持锁状态下调用 waker（close() 是例外，它持着队列锁逐个 wake）——否则 waker 里如果又去 acquire 同一个信号量就会自锁

这是 tokio::sync::Mutex 之所以是公平锁的机械原因。

为什么 `batch_semaphore` 内部要用同步 Mutex

batch_semaphore::Semaphore 用 waiters: Mutex<Waitlist> 保护等待链表。注意这里必须是同步锁（std::sync::Mutex，开了 parking_lot feature 时是 parking_lot::Mutex）——不能用 tokio::sync::Mutex。为什么？

因为循环依赖：tokio::sync::Mutex 内部就是 Semaphore(1)——如果 Semaphore 内部又用 tokio::sync::Mutex，两者互相依赖无法初始化。

同步锁在这里安全的原因：

锁内代码极短（几行链表操作）——微秒内完成
不跨任何 .await——完全同步代码
持锁时间短到不会阻塞 worker 显著时间

这是一个例外——本章前面说"跨 .await 不能用 std Mutex"，但 Tokio 内部自己的实现里用同步 Mutex 保护 hot path 的短操作。例外不矛盾——判断标准是"会不会跨 .await"而不是"能不能用"。

这种 "底层用同步锁做极短操作、上层暴露 async 接口" 的 pattern 在 Tokio 源码里反复出现——Waitlist、OwnedTasks、RegistrationSet 都是这种设计。这些同步锁都是 crate::loom::sync::Mutex：开 parking_lot feature（full 包含）时是 parking_lot::Mutex，否则是 std::sync::Mutex 的薄包装（tokio/src/loom/std/mod.rs:59-70）。正确使用 std 和 tokio 同步原语的关键是理解它们各自的适用边界。

12.5 公平锁 vs 非公平锁的 trade-off

"公平 vs 非公平"是并发锁设计里的永恒争议——两种策略各有适用场景。公平锁按 FIFO 顺序分配——好处是"不会饿死"、坏处是"fast path 不够快"。非公平锁允许"抢占"——刚 release 的锁如果有正在跑的 task 恰好想 acquire、让它直接拿走、不等 FIFO 队列里的等待者——好处是吞吐量极高（减少上下文切换）、坏处是"可能饿死"。Tokio 的 Semaphore 选择FIFO 公平路线——以牺牲极端吞吐量换取可预测性和饿死保护。这种选择反映了 Tokio "面向通用后端服务"的定位。

公平锁（FIFO）：先等的先得。tokio::sync::Mutex 是这种。 非公平锁（barging）：新来的可能"插队"——释放锁的瞬间刚好有线程尝试 acquire、直接拿走。std::sync::Mutex 是这种（Rust 1.62 起 Linux 上直接基于 futex 自研实现，不再包 pthread；第三方库 parking_lot 默认也允许插队，但平均每 0.5ms 强制一次公平交接，官方称为 eventual fairness）。

两者 trade-off：

维度	公平锁	非公平锁
延迟分布	均匀	波动大（有的快有的慢）
吞吐	稍低	稍高
饿死风险	无	有（理论上）
实现复杂度	高	低

Tokio 选公平锁的原因：

热点锁上不断有新 Task 涌入时，允许插队会让早到的 Task 被反复越过，等待时间没有上界
公平锁保证最长等待时间有界——对尾延迟友好
实现上已经有 Semaphore 的等待队列——公平几乎免费

std 选非公平锁的原因：

OS thread 调度已经带"模糊公平"——极少饿死
非公平提供稍高吞吐——通用锁的常见优化
CPU 亲和性：刚释放锁的 CPU 的缓存还热，让它立刻再拿更高效

两种选择没有绝对优劣——各自是所在场景的最优。记住这个定性结论就够了：tokio::sync::Mutex 严格 FIFO、可预测但每次交接都要走一遍唤醒；std::sync::Mutex 允许插队、无竞争时更快但顺序无保证。 至于"慢多少"，取决于竞争程度和持锁时长，只能在你自己的负载上测。

公平锁在尾延迟上的价值

一个公平锁让 p99 延迟更稳定可预测。具体对比：

公平锁：每个等待者的排队位置确定，等待时间有上界（前面还有几个人是可数的）——延迟分布紧凑
非公平锁：一个不断有新请求涌入的热点锁上，早到的等待者可能被反复插队——平均更快、但尾部可以很长

对 SLA 驱动的服务（比如广告竞价必须在 100 ms 内返回响应），p999 比 p50 更重要——公平锁是天然选择。

对吞吐驱动的服务（比如批处理、ETL），总吞吐优先——非公平锁更合适。

Tokio 面向"后端服务"定位——选公平锁符合它的目标用户。这是设计决策和场景匹配的例子。

非公平锁为什么总体更快

你可能好奇：既然非公平锁有"饿死"风险，为什么总体吞吐更高？

因为 CPU cache 热度：刚释放锁的 Task 通常是刚刚访问数据的——它的 L1/L2 cache 是热的。如果让它立刻再拿锁，下次访问依然命中 cache。

如果公平排队让其他 Task 先拿（数据在别的 CPU cache 里——需要 cache 迁移）——每次锁切换伴随一次 cache miss，累积下来吞吐降低。

非公平锁的"插队"其实是 cache-friendly 调度——它把锁给了最可能快速完成的人。代价是偶发的长尾。

理解这个 trade-off 后，你能为自己的场景做出判断——没有绝对答案。

`close` 机制：如何优雅关闭 Semaphore

Semaphore 还提供 close() 方法——标记 Semaphore 为关闭。关闭后：

所有等待中的 acquire().await 返回 AcquireError
新的 acquire 立刻返回 Err
已经持 permit 的 Task 不受影响（它们继续）

用途：服务 shutdown 时优雅地中止所有等待——避免 "shutdown 但有 Task 在等死锁资源"。

let sem = Arc::new(Semaphore::new(10));

// ... 正常业务 ...

// 收到 shutdown 信号
sem.close();
// 所有正在 acquire().await 的 Task 醒来、拿到 Err、走错误处理路径

没有 close 的话，shutdown 期间正在 acquire 的 Task 可能永远等不到 permit（因为不再有 release）——程序 hang 住。

这是一个"想起才会用、不用会栽"的特性。生产服务的 graceful shutdown 设计中必不可少。

Semaphore 的 permit 上限

batch_semaphore 里写着 pub(crate) const MAX_PERMITS: usize = usize::MAX >> 3;（tokio/src/sync/batch_semaphore.rs:130），公开 API 上也有对应的 pub const Semaphore::MAX_PERMITS（tokio/src/sync/semaphore.rs:444）——64 位平台上是个天文数字，足够所有合理场景。超过这个范围，Semaphore::new 会 panic，release 路径则在 add_permits_locked 里 assert 失败。

为什么是 >> 3 而不是 >> 1？实际被占用的只有最低 1 位（const CLOSED: usize = 1; const PERMIT_SHIFT: usize = 1;）。源码注释（tokio/src/sync/batch_semaphore.rs:123-129）给了原因：旧版信号量实现用了 3 个标志位，为了将来需要加标志时不造成破坏性变更，继续保留这 3 位。

注意 RwLock 里另有一个 const MAX_READS: u32 = u32::MAX >> 3;（tokio/src/sync/rwlock.rs:24）——那是"最大并发读者数"，和信号量的 MAX_PERMITS 不是一个东西。

这种 packing 你在整个专栏看过多次——第 6 章 Task State、第 8 章 ScheduledIo readiness、下一章 oneshot 的四个状态位和 watch 的版本号。Tokio 里几乎每个承担状态的 AtomicUsize 都 packing 了多种信息。（第 11 章 Level::occupied 是另一回事——那是纯 bitmap，一位对一个槽，不是多字段 packing。）

一段真实案例：Semaphore 用作限流器

Semaphore 最常见的应用之一是并发限流——控制一个资源（DB 连接、外部 API 调用、CPU 密集任务）的并发度：

use tokio::sync::Semaphore;
use std::sync::Arc;

// 限制并发 10 个
let sem = Arc::new(Semaphore::new(10));

for request in requests {
    let permit = sem.clone().acquire_owned().await.unwrap();
    tokio::spawn(async move {
        let _permit = permit;  // 持 permit 直到 Task 结束
        handle_request(request).await;
        // permit drop 时自动释放
    });
}

5 行代码实现一个工业级并发限流器——每个请求拿一个 permit 才能开工、permit 数满了等释放、drop permit 自动归还。

对比 channel 做限流：你得手动管理 buffered channel、维护 worker loop、处理 backpressure。Semaphore 把这些都省了。

一段真实案例：连接池用 Semaphore 控制容量

数据库连接池也是 Semaphore 的典型应用：

struct Pool {
    sem: Semaphore,        // 限制最大连接数
    conns: Mutex<Vec<Conn>>, // 空闲连接
}

impl Pool {
    async fn acquire(&self) -> PoolConn<'_> {
        let permit = self.sem.acquire().await.unwrap();
        let conn = match self.conns.lock().await.pop() {
            Some(c) => c,
            None => Conn::new().await,  // 空闲列表空、新建
        };
        PoolConn { conn, permit }
    }
}

Semaphore 限制总连接数、Mutex<Vec> 管空闲池——两个基础原语组合出一个完整的连接池。生产级连接池也是这副骨架，只是零件更精细——比如 sqlx 0.8 的池就是一个信号量限总数、一个空闲连接队列（换成了无锁的 ArrayQueue，见 sqlx 0.8.6 PoolInner 的 semaphore 与 idle_conns 两个字段）。

Semaphore 为什么比 channel 合适做并发限流

你可能会问：为什么不用 bounded channel 做限流？比如发送到容量 N 的 channel、channel 满了阻塞发送端——似乎也能限流？

可以，但不推荐，理由：

语义不匹配：channel 本质是"传输数据"、限流是"控制 permit"。两者目的不同、用 channel 实现限流是借用
额外开销：channel 有"数据存储 + 传输路径"，限流场景数据本来就在调用方手里——channel 的存储是浪费
Code 可读性：sem.acquire().await 直接表达"我要一个资源"；tx.send(()).await 需要读者猜"哦这是限流"

API 应该直接表达意图。Semaphore 存在就是为了"并发控制"——别用别的。

12.6 `std::sync::Mutex` 在 async 里能用吗

这是 Rust async 新手最常踩的坑之一。答案是："能用、但要非常小心"。std::sync::Mutex 是阻塞线程的锁——在 async fn 里 lock 它、如果锁被其他线程持有、当前线程（也就是 Tokio worker）会被整个 blocking 住——整个 worker 上的几千个 task 都跑不了。所以规则是：在 async 里用 std Mutex 时、lock 持有时间必须极短（微秒量级）、绝对不能跨 .await。本节把这个规则以及何时应该用 tokio::sync::Mutex 讲清楚。

一个极其常见的问题。答案是：短操作可以、跨 .await 绝对不行。

短操作可以：

let counter = std::sync::Mutex::new(0);
// ...
let mut v = counter.lock().unwrap();   // std Mutex
*v += 1;
drop(v);                                // 立刻释放

同步锁 + 同步操作——极短时间内完成，不阻塞 worker 显著时间。可接受。

跨 .await 绝对不行：

let mut v = counter.lock().unwrap();   // std Mutex
some_io().await;                        // ❌ 跨 .await 了
*v += 1;

先说最常撞上的那一条：std::sync::MutexGuard 是 !Send（library/std/src/sync/poison/mutex.rs:289 的 impl<T: ?Sized> !Send for MutexGuard<'_, T> {}），guard 跨 .await 活着会让整个 Future 变成 !Send，而 tokio::spawn 不论哪种 runtime 都要求 Send，于是直接编译不过——大多数人根本走不到运行期。下面两个后果，是在能编过的写法里（LocalSet 的 spawn_local、block_on 的顶层 future，或压根没 spawn 只是就地 .await）才显现的：

1. worker 被阻塞：some_io.await 如果需要等事件、Task 会被 yield。但此时 Task 还持有 std Mutex——同一 worker 上其他 Task 想拿这个锁就真的被阻塞（std Mutex 是 blocking 锁）。worker 被独占、其他 Task 饿死。

2. 可能真死锁：如果另一个 Task 在同一个 worker 上等同一个 std Mutex，两者会 deadlock。

结论：

纯 sync 短操作 → std::Mutex（快）
跨 .await → tokio::sync::Mutex（必须）
不确定 → tokio::sync::Mutex（安全）

为什么 Clippy 有 `await_holding_lock` lint

Clippy 提供 await_holding_lock lint——检测"持 std Mutex 跨 await"。它默认就是 warn，你写了这种代码，Clippy 会警告。生产代码别把它 allow 掉，修复所有警告。

`parking_lot::Mutex` 在 Rust 生态的位置

parking_lot 是 Rust 生态里一个替代 std 同步原语的实现库。它的卖点是：

锁本身只占 1 字节，Mutex<T> 不额外撑大结构
没有 poisoning——lock() 直接返回 guard，不用 .unwrap()
争用下先自旋一小会儿再 park（adaptive spinning）

需要说明的是，std 这边也在追：Rust 1.62 起 std::sync::Mutex 在 Linux 上换成了直接基于 futex 的自研实现，不再包 pthread mutex，体积和快路径开销都大幅下降。所以"parking_lot 快多少"这个问题，答案强烈依赖你用的 Rust 版本、平台和竞争程度——别照搬任何旧文章里的百分比。

对 async 场景的讨论：

parking_lot 和 std 都是同步锁——同样不能跨 .await 持
短操作场景 parking_lot 可能更快——但对 async 的跨锁讨论 parking_lot 仍然不安全
Tokio 内部的同步锁是 crate::loom::sync::Mutex：开 parking_lot feature（full 会打开它）时是 parking_lot::Mutex，否则是 std::sync::Mutex 的薄包装（tokio/src/loom/std/mod.rs:59-70）——不强制依赖第三方

建议：对非 async 代码用 parking_lot；async 代码里短操作用 std::sync::Mutex、跨 await 用 tokio::sync::Mutex。不要在 async 里为了性能用 parking_lot——因为它仍然是同步锁，同样有阻塞风险。

12.7 RwLock：读写的 permit 共享

Tokio RwLock 的实现利用了 Semaphore 一个精妙扩展：一次 acquire 可以取多个 permit。RwLock 内部是一个拥有 MAX_READS 个 permit 的 Semaphore（常量名就叫 MAX_READS，不是 MAX_READERS，见 tokio/src/sync/rwlock.rs:24）——一个读者 acquire 1 个 permit、一个写者 acquire 全部 MAX_READS 个 permit。这样读者之间不互斥（总 permit 数够）、但写者会被所有读者挡住、读者也被写者挡住。一个 Semaphore + 两种 acquire 模式 = RwLock——这种"用已有原语稍加变形实现新功能"是 Tokio 代码复用的典范。

tokio::sync::RwLock 实现类似 Mutex，但默认编译下是 3 个字段——除了 s: Semaphore 和 c: UnsafeCell<T>，还多一个 mr: u32（max readers）。源码里其实还有一个 resource_span 字段，但它被 #[cfg(all(tokio_unstable, feature = "tracing"))] 挡着，只在开了 tokio_unstable 且启用 tracing feature 时才存在，下面的片段按默认编译给出：

// 来源：tokio/src/sync/rwlock.rs:24 与 tokio/src/sync/rwlock.rs:88
const MAX_READS: u32 = u32::MAX >> 3;

pub struct RwLock<T: ?Sized> {
    // maximum number of concurrent readers
    mr: u32,
    //semaphore to coordinate read and write access to T
    s: Semaphore,
    //inner data T
    c: UnsafeCell<T>,
}

信号量初始化为 Semaphore::new(MAX_READS as usize)，然后：

读锁 → self.s.acquire(1).await —— 占 1 permit
写锁 → self.s.acquire(self.mr as usize).await —— 占 全部 mr 个 permit

（RwLock::with_max_readers(t, n)（tokio/src/sync/rwlock.rs:271）可以把 mr 调小，用来限制同时读的数量。）

语义：

多个读者可同时持锁（每个只拿 1 permit）
写者拿走所有 permit——其他读者和写者都被挡
写者归还 permit 后、等待的读者 / 写者按 FIFO 顺序被唤醒

这套设计的精妙：用同一个 Semaphore 实现读写锁——不需要单独的"读计数 + 写标志"数据结构。permit 数字天然表达"有多少资源"。

写者饥饿的防范

如果读者接连不断来，写者会不会永远凑不齐 mr 个 permit？不会——防饥饿的机制正是 12.4 讲的"batch semaphore 可以部分获取"：

写者 acquire(mr) 时，信号量把当时所有可用的 permit 一次性给它（next = 0），把差额记在 waiter 上继续排队
于是后来的读者调 acquire(1) 时看到 permits == 0，fast path 失败、只能排到写者后面
在读的读者陆续 release(1)，这些 permit 通过 add_permits_locked 直接交接给队首的写者，而不是还回计数器让新读者抢走

"部分获取 + permit 直接交接给队首"这两条合起来，才让写者的等待有上界。代价是一旦有写者在等，新读者即使"逻辑上可以并发读"也要排队——这是公平性换来的。

`downgrade` 和 `upgrade`：读写锁的高阶 API

tokio::sync::RwLock 还提供：

write guard → downgrade → read guard（tokio/src/sync/rwlock/write_guard.rs:376）：不释放锁的前提下从写降到读，让其他读者也能进
read guard → upgrade（未提供直接 API，但可以 drop + re-lock）：从读升到写

downgrade 在 "写完后还要读一会儿、但不想独占" 的场景下有用。减少一次锁获取。

实际生产中这些 API 使用不多——大多数场景一把 Mutex 就够。复杂锁升降机制往往暗示设计有问题——考虑把临界区拆小。

12.8 `Notify`：条件变量的异步版

Notify 是 Tokio 对传统 Condition Variable（条件变量）的异步化重新设计。传统 Condvar 要和 Mutex 配对使用、代码模式固定（while 判断条件 + wait）；Notify 更轻量——它只管"通知"、不和 Mutex 耦合、用户自由决定通知语义（一次通知所有 / 一次通知一个）。这种"解耦 + 更轻量 API"的设计让 Notify 比 Condvar 更通用——可以做条件变量、做信号、做异步 oneshot 的基础、做 queue 的 consumer notification。

Notify 是一个特殊的同步原语——不保护数据，只提供"通知 + 等待"的信号：

let notify = Arc::new(Notify::new());

// Task A：等通知
notify.notified().await;

// Task B：发通知
notify.notify_one();   // 唤醒一个等待者
// 或
notify.notify_waiters();   // 唤醒所有等待者

用途：实现 producer-consumer 的信号、等待某个条件成立（配合外部 Mutex 保护的状态检查）。

内部实现：和 Semaphore 类似但更简单——一个 atomic 计数 + 等待链表。notify_one 从链表头 wake 一个、notify_waiters wake 全部。

Notify 的三种 wake 模式

Notify（tokio/src/sync/notify.rs:201）一共有三个通知方法（notify_last 是 PR #6520 加的，随 tokio 1.38.0 发布）：

notify_one（tokio/src/sync/notify.rs:579）：唤醒等得最久的那一个等待者；如果当时没人等，会存下一个 "permit"，供下一次 notified() 立刻消费
notify_last（tokio/src/sync/notify.rs:592）：唤醒最近才开始等的那一个；没人等时同样存 permit
notify_waiters（tokio/src/sync/notify.rs:662）：唤醒所有当前等待者，且不存 permit——调用之后才 notified() 的不会立刻 Ready

差别核心：notify_one 有存储、notify_waiters 不存。这个 API 细节如果搞错会导致"通知丢失" bug。

实际使用选择：

1 producer + 1 consumer → notify_one
1 producer + N consumer（"状态变了，大家都来看看"）→ notify_waiters
不确定 → notify_one（会存 permit，不容易丢通知，更保守）

注意 notify_waiters 的丢通知风险：它只叫醒"此刻已经在等的人"。经典的错误写法是先检查条件、再 notified().await——两步之间如果发生 notify_waiters，这次通知就丢了。正确做法是先拿到 Notified future（let fut = notify.notified();）再检查条件，最后才 .await。

Notify 和 tokio::sync::watch 的对比

tokio::sync::watch 是另一个"通知式"原语——但它携带值：

let (tx, mut rx) = tokio::sync::watch::channel(0);

tokio::spawn(async move {
    loop {
        rx.changed().await.unwrap();
        let value = *rx.borrow();
        // 读到最新 value
    }
});

tx.send(1).unwrap();

watch 适合 "订阅最新配置 / 状态"——读者只要最新值，不在乎历史。Notify 是 "有事发生了"——不携带具体信息。

两者互补：

配置热更新：watch
新任务到来、需要处理：Notify
精确的 producer-consumer 消息传递：mpsc（下一章）

选对原语 = 一半代码质量。

12.9 一个致命陷阱：跨 .await 持 Tokio Mutex

这个陷阱是 async Rust 最隐蔽的性能杀手——代码编译通过、单机测试没问题、上生产一压就变成尾延迟灾难。原因是：持有 Tokio Mutex 时 .await、Task 让出、Mutex 仍被持有。其他 task 来 acquire 这个 Mutex 会被挂起、排队数可能几百几千、系统吞吐量断崖式下降。这是一个纯运行时现象——编译器和类型系统都抓不出来、只能靠开发者的自觉。本节讲的规则"Mutex 不跨 .await"是所有 Tokio 代码 review 的红线之一。

前面说过跨 .await 不能持 std::Mutex。那 Tokio Mutex 跨 .await 可以吗？

语法上可以（tokio::sync::Mutex 是 async 安全的：它内部就是一个初始许可为 1 的 batch_semaphore::Semaphore——tokio-1.40.0/tokio/src/sync/mutex.rs:133-137 的字段 s: semaphore::Semaphore，tokio-1.40.0/tokio/src/sync/mutex.rs:367 的 Semaphore::new(1)；MutexGuard 定义在 tokio-1.40.0/tokio/src/sync/mutex.rs:151），但实践上极其危险。

看这段代码：

let mutex = Arc::new(tokio::sync::Mutex::new(Data::new()));
let guard = mutex.lock().await;          // 拿锁

fetch_and_update(&guard).await;          // ❌ 跨 .await 持锁
                                         // 如果 fetch_and_update 内部要拿这把锁，deadlock

后果：

性能坍塌：这把锁在 fetch_and_update 期间（可能几十毫秒）一直被 hold——其他 Task 全排队
死锁风险：fetch_and_update 如果直接或间接再调 mutex.lock()——死锁

正确做法一：最小化持锁区间

let data = {
    let guard = mutex.lock().await;
    guard.clone()                        // 取出需要的数据
};                                        // ← 锁在这里释放

fetch_and_update(&data).await;           // 不持锁

正确做法二：读后更新分两步

let id = {
    let guard = mutex.lock().await;
    guard.next_id()
};  // 释放锁

let result = fetch(id).await;  // 耗时操作

let mut guard = mutex.lock().await;
guard.store_result(result);     // 再拿锁写

每一条 "持 Mutex 跨 .await"都应该被 review——是否真的必要、能否拆分？大多数场景可以拆。

`Clippy::await_holding_lock` 管不到 Tokio Mutex

需要说清楚的是：await_holding_lock 检查的是非 async 感知的锁（std::sync::MutexGuard、std::sync::RwLock 的两个 guard、parking_lot 的 guard）——tokio::sync::MutexGuard（tokio-1.40.0/tokio/src/sync/mutex.rs:151）和 tokio::sync::RwLock 的 guard（tokio-1.40.0/tokio/src/sync/rwlock.rs）本来就是设计成可以跨 .await 持有的，不在这个 lint 的范围内。

换句话说："持 tokio Mutex 跨 await"是 lint 抓不到的，只能靠 code review。这也正是它比 std Mutex 的坑更隐蔽的原因——编译器不拦、clippy 不拦、单机测试也不一定暴露。

`Mutex<HashMap>` 之外还有哪些选择

同一个"多 task 共享一个 map"的需求，在 Rust 里至少有五种做法，它们的代价结构完全不同（本机没有可复现的基准环境，这里只给定性的取舍，不给数字——任何跨机器抄来的 ops/s 都不可信）：

实现	读路径	写路径	async 适配
`std::sync::Mutex<HashMap>`	一次原子 + 临界区	同左	不能跨 `.await` 持有
`tokio::sync::Mutex<HashMap>`	走信号量，无竞争时也是几条原子	同左	可跨 `.await`，但见 12.9 的告诫
`RwLock<HashMap>`	读者可并发	写者独占，且会挡住后来的读者	tokio 版可跨 `.await`
`DashMap`（分片锁）	按 key 分片，读写都只锁一个分片	同左	是同步锁，guard 不能跨 `.await`
`arc_swap::ArcSwap<HashMap>`	读几乎不阻塞	每次写要复制整张表	读端无 guard，天然安全

取舍的形状而不是数字：

读远多于写、且写可以整体替换（配置、路由表）→ ArcSwap 这类"读端零锁"方案最合适
读写都频繁、key 分布均匀 → 分片锁（DashMap）把竞争打散
临界区里要 .await → 只能是 tokio::sync 的锁，且要按 12.9 的方式尽量缩短
只是个计数器 → 直接用 AtomicUsize，别用任何锁

凭感觉选数据结构是性能问题的第一来源——真要定量，就在你自己的负载上测，别抄别人的表。

Mutex 的持有时间 vs 吞吐

一个反直觉的事实：让 Mutex 内的代码极短反而可能降低吞吐。为什么？

极短代码 → Mutex 频繁 acquire / release → atomic 操作和 cache line bouncing 密集
临界区稍长一点 → 每次 acquire 摊到更多有效工作 → 加解锁本身的占比下降

但锁持有时间太长又会导致争用——延迟飙升。

经验法则：临界区既不要短到和一次加解锁本身同量级，也不要长到别人明显在排队。中等粒度往往更好。

具体建议：

如果临界区短到和一次加解锁本身同量级：考虑把相邻的几次操作合并进同一次 lock
如果临界区长到能明显看到别人在排队：考虑拆成多次 lock，中间留出让别人进来的间隙

这类"中等粒度优于极粗和极细"的直觉是经验积累——具体阈值必须在你自己的项目上 profile 出来，任何写死的纳秒数都不可迁移。

12.10 开销结构和选型指南

本节把前面所有同步原语的代价结构 + 适用场景整理成一张对比表——让你在遇到"这里该用什么"时能快速决策。重点记住一个定性结论：tokio::sync::Mutex 的每一次加锁都要走一遍信号量的状态机，竞争时还要经过"注册 waker → 让出 → 被 wake → 重新被调度"的完整往返，而 std::sync::Mutex 无竞争时只是一两条原子指令。所以选型原则是：只在"必须跨 .await 持锁"时才用 tokio::sync::Mutex，其余场景用 std::sync::Mutex + 尽量短的临界区。

代价结构对比（不给绝对耗时——它随平台和竞争程度剧烈变化）：

操作	`tokio::sync::Mutex`	`std::sync::Mutex`
无竞争 acquire	信号量 CAS（`compare_exchange` 循环），纯用户态	一次原子 CAS，纯用户态
有竞争 acquire	拿队列锁 → 挂 waiter → 返回 Pending、让出 worker → 被 wake → 重新排队被调度	线程进内核 futex 等待队列，整个线程挂起
release	拿队列锁 → 把 permit 直接交接给队首 → 批量 wake	一次原子 + 可能一次 futex wake
跨 `.await` 持锁	语法合法，但会把锁按"业务耗时"而不是"临界区耗时"占住（见 12.9）	绝对禁止——会阻塞整个 worker

最关键的差别不在纳秒数，而在"竞争时谁被挂起"：tokio 挂起的是一个 Task，worker 继续跑别的活；std 挂起的是整个 OS 线程，那个 worker 上的所有 Task 一起停摆。

选型速查：

对 HashMap / Vec 等短操作：std::sync::Mutex（快）
跨 async I/O 的共享状态：tokio::sync::Mutex
读多写少：tokio::sync::RwLock 或 arc_swap::ArcSwap
简单共享计数：std::sync::atomic::AtomicUsize（别用 Mutex）
限流 / 并发控制：tokio::sync::Semaphore
通知 / 等待信号：tokio::sync::Notify

12.11 一些被大家忽略的 API

除了 Mutex / RwLock / Semaphore / Notify，tokio::sync 还有几个稍冷门但实用的原语：

`OnceCell<T>` —— 异步版一次初始化

对应 std::sync::OnceLock，但初始化函数可以是 async：

let cell = OnceCell::new();
let value = cell.get_or_init(|| async {
    expensive_async_computation().await
}).await;

适合"懒加载异步资源"——比如启动时拉配置、建立 DB 连接池、初始化 HTTP client。

`broadcast::channel` —— 多消费者广播

多个 sender、多个 receiver，每个 receiver 都能收到订阅之后发出的每条消息（落后太多会收到 Lagged）。底层是 ring buffer——下一章详讲。

`oneshot::channel` —— 单次发送

一次性 send 一个值、接收端收到。用于"异步函数返回值"或"事件通知"。

`Barrier` —— N 个 Task 等齐

N 个 Task 各自 wait()、第 N 个到齐时所有人继续。适合需要协调多个 Task 到达同一阶段的场景（并发压测、并发采集等）。

`Mutex::lock_mut` 的错觉

没有 Mutex::lock_mut——因为 Rust 的所有权让 &Mutex<T>.lock() 返回的 Guard 已经提供可变访问（通过 DerefMut）。命名上的一致性反映语义的一致性。

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

本章讲的 "Mutex = Semaphore(1) 的共通设计" 和 《Vue 3 设计与实现》第 4 章（@vue/reactivity 源码深度剖析（上）：reactive / ref / track / trigger / computed） 里讲的 Vue 响应式节点 = "订阅者链表 + 标志位"——同构思路：用一个通用原语实现所有特例。Vue 里 ref、computed、effect 和 reactive 属性的 Dep 都实现同一个 ReactiveNode 接口、共用 link / propagate 引擎；Tokio 里 Semaphore 是 Mutex / RwLock 的基础。"用少数原语实现多种功能"是好库设计的标志。

《Rust 编译器与运行时揭秘》第 12 章（unsafe：安全抽象的逃生舱） 里讨论的 UnsafeCell 的使用规则——本章的 Mutex.c: UnsafeCell<T> 是典型应用。UnsafeCell 让 T 可以在 &Mutex<T> 的引用下被可变访问——配合 Semaphore 的互斥保证、整体安全。这种"unsafe 内部 + 用 semaphore 保证正确性"是 Rust 低层库的常见模式。

12.13 `Notify` 的"已通知但无等待者"设计

Notify 有一个精细设计：如果没人等时 notify_one()，Notify 会"记住"这次通知。等下次有 Task 调 notified().await 时，它不会挂起，立刻被那次保存的通知满足。

let n = Notify::new();
n.notify_one();                  // 当前没人等——notification 被"存起来"
n.notified().await;              // 立刻返回，消耗那个 notification

这种"存一个未消费通知"防止"通知丢失"：producer 先发、consumer 后等——仍然能被唤醒。

但只存一个——多次 notify_one 不累积：

n.notify_one();
n.notify_one();
n.notify_one();
// 只存一个 notification——第一次 notified().await 消耗掉后其他两次丢失

这和 Go channel / Rust std::mpsc 的语义不同——它们会累积 buffer。Notify 不是消息队列，它是"最多一个 pending 信号"的简单原语。

实用建议：Notify 适合"有新事件需要处理"这类 boolean 信号，不适合"我要送 N 个消息给 consumer"（那是 channel 的工作）。

12.14 RwLock 的读写饥饿陷阱

前面提过 Tokio RwLock 防止写者饥饿——但读者饥饿也可能发生。

假设 3 个读 Task 正在读，这时来了 1 个写 Task，紧接着又来 1 个读 Task：

时间:       0ms  10ms  20ms  30ms  40ms
读 Task A:  R-------
读 Task B:  R------------
读 Task C:  R------------------
写 Task D:    等...等...等...W-------   ← D 到达时 A/B/C 已在读，要等它们全部结束
读 Task E:      等...等...等...等...等D完了才能R   ← E 晚于 D 到达，FIFO 排在 D 后面

如果 D 之后还有连续的读请求——每个读请求都要等 D 先完成（FIFO）。D 如果耗时 5 秒，所有后续读都被 D 挡 5 秒。

这是公平 RwLock 的代价——避免写者饿死、但可能让短读请求等久。

几种缓解办法：

如果读远多于写，考虑 arc_swap::ArcSwap —— 完全 lock-free 读、原子替换写
如果写很少且快，评估要不要用 Mutex（简单模型常常更好）
如果需要"写优先"或"读优先"的复杂策略——自己实现（极少数场景）

"默认用 Mutex，除非测出读写锁有明显收益" 是 Rust 生态的常见建议——RwLock 的 overhead 和微妙语义往往不值得。

12.15 `OwnedMutexGuard` 与跨 async task 传递

tokio::sync::Mutex::lock_owned() 返回 OwnedMutexGuard<T>——持有 Arc<Mutex<T>> 而非引用。这个 Guard 可以 spawn 到另一个 Task：

let mutex = Arc::new(Mutex::new(Data::new()));

let guard = mutex.clone().lock_owned().await;

// 可以 spawn——guard 是 'static
tokio::spawn(async move {
    // 在新 Task 里操作 guard
    process(&guard).await;
});

用 MutexGuard<'a> 做不到——它带 'a 生命周期，不能跨 spawn。OwnedMutexGuard<T> 内部持有 Arc，'static 成立，可以自由 spawn。

代价：每次 lock_owned 都 clone 一次 Arc——多一次原子加。对 hot path 有影响，但对"需要跨 spawn 持锁"的场景是唯一方案。

Semaphore 也有类似的 acquire_owned——返回 OwnedSemaphorePermit，同样可以跨 spawn。

这类 "owned variant" 的 API 是 Tokio 对 Rust 生命周期系统的妥协——类型系统严格时，需要多一个变体让 owned 路径可用。有点冗余但实用。

12.16 DashMap vs tokio::sync::`Mutex<HashMap>`

一个常见选择题：并发 HashMap 用什么？

选项 A：Mutex<HashMap>

优点：简单、标准
缺点：所有操作串行——单写会把读也 block

选项 B：RwLock<HashMap>

优点：多读并发
缺点：写时阻塞所有读、Tokio RwLock overhead 不小

选项 C：DashMap（第三方库）

优点：分片（每 shard 一个 RwLock）——大幅降低锁竞争
缺点：不是 async 安全的——DashMap 的分片锁是同步读写锁、跨 await 持 Guard 会 block

选项 D：Mutex<HashMap> 的 "clone-on-read" 模式

拿锁、clone 出需要的条目、释放锁、在 clone 上操作
适合"读频次高、值小"

选项 E：ArcSwap<HashMap>

写时整个 HashMap 替换
读完全 lock-free
写不频繁时最优

默认从 A（Mutex<HashMap>）起步，压测发现瓶颈再优化。不要过早用 DashMap——DashMap 在 async 场景下的正确性需要你自己保证（guard 不跨 await）。

12.17 为什么 Tokio 不提供 "Condvar"

std::sync::Condvar 是经典的条件变量——搭配 Mutex 使用。Tokio 不提供对应的异步版 Condvar。

Notify + Mutex 的组合可以覆盖 Condvar 的用例：显式的 wait + notify + 循环检查比 Condvar 的"原子 unlock+wait+relock"更清晰——只要守住 12.8 那条顺序（先拿到 Notified future 再检查条件），就不会丢通知。

"不增加 API" 也是一种设计美德——有时候不提供某个 API 比提供更好。学会这种 "克制"是成熟库作者的标志。

12.18 一个真实案例：如何修一个 "跨 .await 持锁" bug

我在 2022 年排查过一个生产 bug，现象：服务在高峰期偶尔 p99 飙到几秒（平时几十毫秒）。

调查后发现一段"无辜"的代码：

pub async fn process_events(&self, events: &[Event]) {
    let mut state = self.state.lock().await;   // tokio::sync::Mutex
    for event in events {
        state.process(event);
        self.audit_log.append(event).await;   // ❌ 跨 .await 持锁
    }
}

audit_log.append 涉及 disk I/O、最坏 20 毫秒。一个 process_events 带 10 个 event → 持锁最长 200 毫秒。期间其他所有需要 state 锁的 Task 全排队——尾延迟爆炸。

修复：缩小锁范围——锁内只做同步的 process，审计日志挪到锁外

{
    let mut state = self.state.lock().await;
    for event in events {
        state.process(event);                // 整批处理仍在同一次持锁内
    }
}                                            // 锁在这里释放
for event in events {
    self.audit_log.append(event).await;      // 锁外逐条写审计
}

这样"先处理、后记审计"的顺序和整批处理的原子性都没变，变的只是 I/O 不再占着锁。

修复后 p99 从几秒降到几十毫秒——一个锁范围改小就解决了。

这类 bug 很隐蔽：

代码逻辑是正确的
单元测试全通过
低负载下跑得很快
只在高并发 + I/O 慢时才暴露

预防方法：

别指望 Clippy 的 await_holding_lock——它只管 std / parking_lot 的锁，抓不到 tokio Mutex（见 12.9）
Code review 时专门检查 "持锁范围"
压测——尤其是带网络延迟的真实环境

FIFO 公平性意味着没人能绕过这个持锁者——某个持锁很久的 Task 会让排在后面的所有 Task 依次等待，排队时间一路累积到尾延迟上。

12.19 本章小结

本章讲了 Tokio sync 模块的核心锁与信号原语——Mutex、RwLock、Semaphore、Notify，并简单提到 Barrier、OnceCell、oneshot、broadcast（channel 留到下一章）。关键 take-home：Mutex / RwLock / OnceCell / bounded mpsc 都建立在 batch_semaphore 之上（Notify、watch、broadcast 走的是各自的等待链表）；公平 vs 非公平是根本取舍；std Mutex 在 async 里有限使用、不跨 await；tokio::sync::Mutex 只在必须跨 .await 持锁时才用。

带走三件事：

Mutex = Semaphore(1) + UnsafeCell<T>——lock / try_lock / Drop 三段核心逻辑加起来十几行有效代码，胜过几千行各自独立的实现。RwLock 只是把它换成 Semaphore(MAX_READS) + "读拿 1、写拿全部"
batch_semaphore 的 fast path 是无锁 CAS、slow path 才进同步 Mutex 保护的 waitlist——无竞争时根本不碰那把锁。而且 release 是把 permit 直接交接给队首等待者，剩余的才还回计数器
Tokio Mutex 是公平锁（FIFO 等待队列）——保证最大等待时间有界、适合延迟敏感场景。公平性的代价不是"慢多少倍"这种固定数字，而是每次锁交接都必须走一遍完整的唤醒 + 重新调度

一个升华

读完本章，你应该意识到：Tokio 的所有复杂同步行为都建立在 "permit + wait queue + atomic" 这三个原语上。不同原语是这三件的不同组合：

Mutex = 1-permit + FIFO queue
RwLock = N-permit + FIFO queue + 读写分配
Semaphore = N-permit + FIFO queue
Notify = boolean permit + FIFO queue
Barrier = countdown permit + broadcast

学会"拆解任何同步原语到这三件基础物料"—— 你以后看任何并发库（Java concurrency、Go runtime sync、C++ 标准库的 <mutex> / <condition_variable>）都有了统一视角。这种基础物料的识别能力是工程师最珍贵的心智工具之一。

上下游对接：这些原语如何流到第 13 章

下章讲 channel（mpsc / broadcast / watch / oneshot）——你会发现这些 channel 的 backpressure / blocking 机制都能在本章找到原型（只有 bounded mpsc 直接用了 Semaphore）：

bounded mpsc 的发送端：满时用 Semaphore 的 acquire() 限流
broadcast：Box<[RwLock<Slot<T>>]> 环形缓冲 + 自己的等待链表（不走 Semaphore、也不走 Notify），落后的接收端会收到 RecvError::Lagged
watch：RwLock<T> 存最新值 + AtomicState 存版本号（最低位当 closed 标志——又一次 bit packing），唤醒订阅者用的是包在 Notify 之上的 BigNotify
oneshot：一个 AtomicUsize 状态字 + UnsafeCell<Option<T>>，两端各自的 waker——完全自成一套

permit + 等待队列 + 原子状态字——这三样物料的不同配比组合出整个 channel 家族。学会本章，下章几乎就是变体欣赏。

读完本章你应该能回答的几个问题

离开本章前，用这几个问题自测你的理解：

问题 1：Tokio Mutex 持锁时其他 Task 怎么办？答：被放进 Semaphore 的 waitlist、返回 Pending、让出 worker。worker 继续跑别的 Task。

问题 2：std Mutex 和 tokio Mutex 底层分别怎么挂起？答：std Mutex 让 OS 线程进 kernel wait（futex）。tokio Mutex 让 Task 挂起（Waker 机制）——OS 线程不被挂起。

问题 3：RwLock 为什么 Tokio 默认公平？答：避免写者饥饿。底层 Semaphore 的 FIFO waitlist 天然支持。

问题 4：Semaphore 的 fast path 是什么？答：permits 足够时直接 CAS 扣减——零锁、零等待。只有不够才去拿 waitlist 的同步 Mutex。

问题 5：为什么不能在 .await 跨 std Mutex？答：std Mutex 阻塞 OS 线程 —— 同线程其他 Task 被卡 —— 可能死锁（下一个 Task 也要这把锁时）。

如果这 5 题你都能答清，本章的核心你已经掌握。

一个加分题：Arc<Mutex<T>> 真的需要 Arc 吗？能不能只用 &Mutex<T>？

答：有时不需要 Arc——如果 Mutex 的生命周期明显超过所有使用点（单线程 runtime 栈上的 Mutex），&Mutex<T> 就够。Arc 只在"需要跨 'static 边界"（spawn 到其他 Task）时才必要。

很多 Rust 新手下意识写 Arc<Mutex<T>>——其实不必要的情况很多。学会识别"真的需要 Arc vs 只是习惯"是一个小但有用的技能。

从同步原语到 channel：一个心智模型的迁移

下一章讲 channel——你需要的前置知识大部分在本章建立。具体迁移：

Semaphore acquire/release → mpsc 的 permit 控制
Mutex 的 critical section → channel 内部对 ring buffer 的修改
Notify 的 wake → channel 的 receiver 被唤醒
FIFO waitlist → channel 的发送者排队

channel 是这些原语的组合使用——不是全新机制。读下一章时带着"这是哪个原语的应用"的视角，代码读起来会极其流畅。

Semaphore 与 channel 的神秘亲属关系

细心的读者会发现：Semaphore 和 bounded channel 在本质上几乎同构：

bounded channel 满 → send 阻塞 = Semaphore permits = 0 → acquire 阻塞
bounded channel 空 → recv 阻塞 = 反向 Semaphore permits = 0 → acquire 阻塞
channel 消费一个 → Semaphore release(1)
channel 新增一个 → Semaphore acquire(1)

事实上，bounded mpsc 内部用了 Semaphore 做 permit 管理——send 前先 acquire 一个 permit、recv 后 release。这就是"一个基础原语撑起多个 API"的威力。下章讲 channel 时你会看到具体代码。

识别"同构性" 是理解复杂系统的关键——看似不同的 API 往往共用底层机制。识别之后你不再需要单独学每个 API——学会基础原语、其他自然理解。

下一章进入 channel——Tokio 的消息传递原语。你会看到 mpsc 的分块链表队列、broadcast 的环形缓冲、oneshot 的单发设计、watch 的"最新值订阅"——每一个都是同步原语的经典变体。你会明显感受到本章建立的基础在下一章开始结出果实——一旦你已经理解 Semaphore 和 Notify，channel 的实现变得极其清晰。

延伸阅读

Tokio 源码：tokio/src/sync/mutex.rs
Tokio 源码：tokio/src/sync/semaphore.rs
Tokio 源码：tokio/src/sync/batch_semaphore.rs
Clippy lint: await_holding_lock
《Vue 3 设计与实现》第 4 章：通用原语（ref / track / trigger）实现多种功能的设计哲学对比
《Rust 编译器与运行时揭秘》第 12 章：UnsafeCell 与内部可变性的 unsafe 边界

12.1 一个直觉问题：async 需要自己的 Mutex 吗

12.2 Tokio 同步原语全景

12.3 Mutex 的惊人简单

new 的实现

lock 的实现

try_lock 的非阻塞尝试

MutexGuard 的 Drop