Rust in Everything
基于 tokio 1.54.1

tokio::sync::Mutex 还是 std::sync::Mutex

主线位置：主线之外。tokio::sync::Mutex 是一个许可数为 1 的信号量，等锁的任务挂在信号量的等待队列上。

flowchart TB
  q{锁住期间要 .await 吗}
  q -->|不要| std["std::sync::Mutex"]
  q -->|要| r{保护的是什么}
  r -->|"IO 资源，如连接"| t1["tokio Mutex 或独立任务"]
  r -->|普通数据| t2[改代码，别跨 await 持锁]
  std --> wrap[藏在同步方法后面]

tokio 文档怎么说

tokio::sync::Mutex 的文档有一节「Which kind of mutex should you use?」：

Contrary to popular belief, it is ok and often preferred to use the ordinary Mutex from the standard library in asynchronous code.

The feature that the async mutex offers over the blocking mutex is the ability to keep it locked across an .await point. This makes the async mutex more expensive than the blocking mutex, so the blocking mutex should be preferred in the cases where it can be used. The primary use case for the async mutex is to provide shared mutable access to IO resources such as a database connection. If the value behind the mutex is just data, it's usually appropriate to use a blocking mutex such as the one in the standard library or parking_lot.

Note that, although the compiler will not prevent the std Mutex from holding its guard across .await points in situations where the task is not movable between threads, this virtually never leads to correct concurrent code in practice as it can easily lead to deadlocks.

A common pattern is to wrap the Arc<Mutex<...>> in a struct that provides non-async methods for performing operations on the data within, and only lock the mutex inside these methods.

Additionally, when you do want shared access to an IO resource, it is often better to spawn a task to manage the IO resource, and to use message passing to communicate with that task.

归纳成一句：默认用 std 的锁；锁要跨 .await 持有时才用 tokio 的；如果保护的是 IO 资源，还可以考虑用一个独立任务加 channel。

📝NOTE
冷知识：「Which kind of mutex」这一节是 0.2.22 才有的

在那之前，文档说的是：锁不需要跨 await 持有、每次持有接近瞬时，这两种情况应该用标准库的锁；「另一方面，tokio 的锁用于需要持有更长时间、或者跨 await 持有的情况」。#2658（0.2.22，2020 年 7 月）把它换成了「Which kind of mutex should you use?」一节，开头就是「Contrary to popular belief」，「持有更长时间」不再算选 tokio 锁的理由，还加上了「藏在同步方法后面」和「用独立任务管理 IO 资源」两种写法。

#3645（1.5.0）又补了两处：tokio 的锁更贵，所以能用 std 的地方就用 std；以及上面引文里「编译器不会阻止……几乎从来得不到正确的并发代码」那一段。PR 说明的理由是，原来的开头可能被读成「std 的锁不能跨任何 .await 持有」，而事实并非如此：在不会换线程的任务里编译器是放行的，本页后面 spawn_local 的死锁例子就是这种情况。

两把锁的差别

std::sync::Mutextokio::sync::Mutex
拿不到锁时阻塞线程任务挂起，线程去跑别的任务
guard 能否跨 .await（在 tokio::spawn 里）不能，guard 不是 Send能
公平性不保证严格 FIFO
加锁、解锁时的额外工作无加锁要过一次协作预算检查；解锁每次都拿信号量等待队列的内部锁
持锁时 panic锁被标记为 poisoned不 poison，直接解锁

后两行的出处：Mutex::lock 等的是信号量的 Acquire future，它的 poll 拿到许可后还要过一次 coop::poll_proceed（见协作式调度预算）；guard 的 drop 调 Semaphore::release：

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());
}

不管有没有人在等，都先锁住等待队列，把许可按 FIFO 分给排队的任务。信号量的结构和等待队列见 07 同步原语的 batch_semaphore 一页。

📝NOTE
没有蠢问题

问： tokio 的锁内部自己就有一把阻塞锁（等待队列的锁），在 async 代码里用阻塞锁不是不好吗？

答： 这正是上面文档推荐的那种用法。Semaphore 的 waiters 是 crate::loom::sync::Mutex：开了 parking_lot feature（full 包含它）就是 parking_lot::Mutex，否则是 std::sync::Mutex 外面包一层、去掉了 poison。它只在改等待链表、分配许可的那几行同步代码里持有，从不跨 .await。「async 里别用阻塞锁」说的是临界区长、或者要跨 .await 持有的情况。

实测开销

无竞争：一个任务连续加锁、加一、解锁 1000 万次。有竞争：多线程运行时 4 个 worker，8 个任务各做 20 万次同样的操作。

use std::hint::black_box;
use std::sync::Arc;
use std::time::Instant;

const N: u64 = 10_000_000;
const TASKS: u64 = 8;
const PER_TASK: u64 = 200_000;

fn main() {
  // Uncontended: one task, lock + increment + unlock, N times
  let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
  rt.block_on(async {
    let m = std::sync::Mutex::new(0u64);
    let t = Instant::now();
    for _ in 0..N {
      *black_box(&m).lock().unwrap() += 1;
    }
    let std_ns = t.elapsed().as_nanos() as f64 / N as f64;

    let m = tokio::sync::Mutex::new(0u64);
    let t = Instant::now();
    for _ in 0..N {
      *black_box(&m).lock().await += 1;
    }
    let tokio_ns = t.elapsed().as_nanos() as f64 / N as f64;
    println!("uncontended, 1 task:   std {std_ns:.1} ns/op   tokio {tokio_ns:.1} ns/op");
  });

  // Contended: 8 tasks on 4 workers hammer one counter with tiny critical sections
  let rt = tokio::runtime::Builder::new_multi_thread().worker_threads(4).build().unwrap();
  rt.block_on(async {
    let m = Arc::new(std::sync::Mutex::new(0u64));
    let t = Instant::now();
    let hs: Vec<_> = (0..TASKS)
      .map(|_| {
        let m = m.clone();
        tokio::spawn(async move {
          for _ in 0..PER_TASK {
            *m.lock().unwrap() += 1;
          }
        })
      })
      .collect();
    for h in hs {
      h.await.unwrap();
    }
    let std_ns = t.elapsed().as_nanos() as f64 / (TASKS * PER_TASK) as f64;
    assert_eq!(*m.lock().unwrap(), TASKS * PER_TASK);

    let m = Arc::new(tokio::sync::Mutex::new(0u64));
    let t = Instant::now();
    let hs: Vec<_> = (0..TASKS)
      .map(|_| {
        let m = m.clone();
        tokio::spawn(async move {
          for _ in 0..PER_TASK {
            *m.lock().await += 1;
          }
        })
      })
      .collect();
    for h in hs {
      h.await.unwrap();
    }
    let tokio_ns = t.elapsed().as_nanos() as f64 / (TASKS * PER_TASK) as f64;
    assert_eq!(*m.lock().await, TASKS * PER_TASK);
    println!("contended, 8 tasks/4w: std {std_ns:.1} ns/op   tokio {tokio_ns:.1} ns/op");
  });
}

连跑三次：

uncontended, 1 task:   std 6.8 ns/op   tokio 21.8 ns/op
contended, 8 tasks/4w: std 18.7 ns/op   tokio 100.9 ns/op
uncontended, 1 task:   std 6.8 ns/op   tokio 21.8 ns/op
contended, 8 tasks/4w: std 17.6 ns/op   tokio 98.8 ns/op
uncontended, 1 task:   std 6.8 ns/op   tokio 21.8 ns/op
contended, 8 tasks/4w: std 17.8 ns/op   tokio 99.4 ns/op

无竞争时 tokio 的锁约慢 3 倍，竞争时约慢 5 倍。竞争时，tokio 的锁按 FIFO 把许可直接交给排队的任务，每次交接都要唤醒对方、等调度器再 poll 它；std 的锁没有这一步，等锁的线程被唤醒后自己去抢。

几十纳秒和一次网络 IO 比可以忽略，所以这不是「tokio 的锁不能用」，而是「没有理由的时候别用」。

std 锁的常用写法

文档里说的模式：把 Arc<Mutex<..>> 包进一个结构体，只提供同步方法，锁只在方法内部拿。guard 出不了方法，也就不可能跨 .await：

#[derive(Clone, Default)]
struct Db {
  inner: Arc<Mutex<HashMap<String, u64>>>,
}

impl Db {
  fn incr(&self, key: &str) -> u64 {
    let mut map = self.inner.lock().unwrap();
    let v = map.entry(key.to_string()).or_insert(0);
    *v += 1;
    *v
  }
}

关键是临界区短：只做内存操作，不做 IO，不做大计算。临界区长了，lock() 等锁时阻塞的是 worker 线程，就变成了在 async 里阻塞。

什么时候需要跨 .await 持锁

锁保护的对象本身要在锁里做异步操作，而且这几个操作之间不能被别人插进来。比如一条连接，一个任务要连续写两条消息，中间不能夹别的任务的消息：

struct Conn {
  sent: Vec<String>,
}

impl Conn {
  async fn send(&mut self, msg: &str) {
    tokio::time::sleep(Duration::from_millis(5)).await; // stands for a socket write
    self.sent.push(msg.to_string());
  }
}

// three tasks, each holding the lock across two sends
let conn = Arc::new(tokio::sync::Mutex::new(Conn { sent: Vec::new() }));
let hs: Vec<_> = (0..3)
  .map(|i| {
    let conn = conn.clone();
    tokio::spawn(async move {
      let mut c = conn.lock().await;
      c.send(&format!("{i}-a")).await; // a and b of one task stay adjacent
      c.send(&format!("{i}-b")).await;
    })
  })
  .collect();

Db pattern: hits = 4000
tokio Mutex across await: ["0-a", "0-b", "1-a", "1-b", "2-a", "2-b"]
blocking_lock from spawn_blocking: 6 messages

（第一行是 4 个任务各调 1000 次 Db::incr 的结果，第三行见下文 blocking_lock。）

这种场景也可以换成文档建议的另一种写法：起一个任务独占连接，别的任务通过 mpsc 把消息发给它。没有锁，顺序由 channel 保证。

用 std 锁跨 .await 会怎样

在 tokio::spawn 里编译不过（见 future cannot be sent between threads safely）。在 spawn_local 里能编译，但会死锁：

#[tokio::main(flavor = "current_thread")]
async fn main() {
  std::thread::spawn(|| {
    std::thread::sleep(Duration::from_secs(2));
    println!("watchdog: still stuck after 2s, exiting");
    std::process::exit(1);
  });

  let m = Rc::new(Mutex::new(0));
  let local = tokio::task::LocalSet::new();
  let m1 = m.clone();
  local.spawn_local(async move {
    let mut g = m1.lock().unwrap(); // compiles: spawn_local does not need Send
    println!("A: locked, now awaiting while holding the std guard");
    tokio::time::sleep(Duration::from_millis(10)).await;
    *g += 1;
    println!("A: done");
  });
  let m2 = m.clone();
  local.spawn_local(async move {
    println!("B: trying to lock");
    *m2.lock().unwrap() += 1; // blocks the only thread; A can never resume
    println!("B: done");
  });
  local.await;
}

A: locked, now awaiting while holding the std guard
B: trying to lock
watchdog: still stuck after 2s, exiting

A 拿着锁挂起，B 在同一个线程上阻塞等锁，A 需要这个线程才能恢复。std 文档对同一线程重复加锁的说法是「可能死锁，也可能 panic」，这台机器（macOS）上是死锁。

blocking_lock

同一把 tokio 锁既要在 async 代码里用、又要在同步代码里用时，同步一侧调 blocking_lock：

let c2 = conn.clone();
let n = tokio::task::spawn_blocking(move || c2.blocking_lock().sent.len()).await.unwrap();

它就是 crate::future::block_on(self.lock())，只能在没进入运行时的线程上调用（spawn_blocking 的闭包、普通线程）。在 async 代码里调用会 panic Cannot block the current thread from within a runtime，见 Cannot start a runtime from within a runtime。

选哪个

情况选择
保护普通数据（计数器、HashMap、配置），临界区里没有 .awaitstd::sync::Mutex（或 parking_lot::Mutex），藏在同步方法后面
锁里要做异步 IO，几步之间不能被打断tokio::sync::Mutex
共享一个 IO 资源，访问很频繁一个独占它的任务 + mpsc，常常比锁更简单
读多写少的数据std::sync::RwLock；要跨 .await 才用 tokio::sync::RwLock
同步代码要拿 tokio 的锁blocking_lock，只在运行时之外的线程上
临界区很长（大计算、同步 IO）先把长操作移出临界区；移不出来就整个放进 spawn_blocking

小结

  • tokio 文档的建议：默认用 std 的锁，只在锁要跨 .await 持有时用 tokio 的。
  • tokio 的锁是许可数为 1 的信号量：FIFO、加锁要过协作预算、解锁每次都拿内部锁。实测无竞争约慢 3 倍，竞争约慢 5 倍。
  • std 锁的常用写法是把锁藏在同步方法后面，guard 出不了方法；临界区要短。
  • std 的 guard 跨 .await 在 tokio::spawn 里编译不过，在 spawn_local 里能编译但会死锁。
  • blocking_lock 给同步代码用，在 async 里调用会 panic。

💬 关联讨论

发起讨论
加载中...