并发与异步编程

📚 Rust 课程系列
课程概览
基础语法
函数与输入输出
所有权、借用与生命周期
结构体、枚举与模式匹配
类型系统：泛型、trait 与多态
集合与容器
错误处理与 Panic 恢复
模块、属性与宏
智能指针、迭代器与闭包
并发与异步编程（本文）
Unsafe Rust 与常用 trait 详解
工具链、Cargo 与外部 crate
最佳实践、性能与调试

Rust 的并发安全由类型系统在编译期保证：Send/Sync 标记 trait 让「能否跨线程传递/共享」成为类型的一部分，从而「无畏并发」（fearless concurrency）。本章覆盖基于线程的并发（std::thread、通道、Mutex/RwLock/Condvar、线程本地存储、scope 线程）与基于 async/.await 的异步并发（Future、运行时、join!/select!）。

并发

Send 与 Sync

trait	含义	自动推导规则	反例
`Send`	`T` 的所有权可安全跨线程转移	所有字段均为 `Send`	`Rc<T>`、`*const T`
`Sync`	`&T` 可安全跨线程共享	所有字段均为 `Sync`（等价于 `&T: Send`）	`Cell<T>`、`RefCell<T>`

💡 提示：Send/Sync 是标记 trait（marker trait），没有方法，仅作为编译器的「通行证」。手动 impl Send / impl Sync 是 unsafe 的——你在向编译器担保安全性。

🔄 对比：C++ 的线程安全靠程序员自觉（std::mutex 配对靠约定），Rust 则在编译期拒绝不安全的跨线程访问——Rc 不是 Send，编译器直接报错，而非运行时崩溃。

线程与消息传递

use std::thread;
use std::sync::mpsc;

let (tx, rx) = mpsc::channel();
let handle = thread::spawn(move || {
    tx.send(42).unwrap();  // move tx into thread
});
println!("Received: {}", rx.recv().unwrap());
handle.join().unwrap();

💡 提示：thread::spawn 要求闭包 'static，因此捕获的局部变量必须 move 进线程。若需借用栈上数据，见下方 Scope 线程。

💡 提示：mpsc 适合生产者-消费者模式；对性能敏感场景可用 crossbeam 提供更高性能的多生产者通道。

共享状态：Mutex 与 Arc

use std::sync::{Arc, Mutex};
let counter = Arc::new(Mutex::new(0));
let c2 = Arc::clone(&counter);
let handle = std::thread::spawn(move || {
    *c2.lock().unwrap() += 1;  // lock() returns MutexGuard (RAII)
});
handle.join().unwrap();
println!("{}", *counter.lock().unwrap()); // 1

⚠️ 注意：若只是计数，优先使用 AtomicUsize 等原子类型，避免锁开销。Mutex 适用于保护复杂共享数据。

⚠️ 注意：避免在持锁期间调用外部代码或再加锁——这是死锁的经典来源。尽量缩小锁的持有范围。

RwLock

use std::sync::RwLock;
let lock = RwLock::new(5);
{ let r1 = lock.read().unwrap(); let r2 = lock.read().unwrap(); } // 多读者并发
{ *lock.write().unwrap() += 1; }  // 独占写

⚠️ 注意：std::sync::RwLock 在写饥饿场景下可能表现不佳；对性能敏感场景考虑 parking_lot::RwLock（写优先，更小的内存开销）。

Condvar

use std::sync::{Arc, Mutex, Condvar};
let pair = Arc::new((Mutex::new(false), Condvar::new()));
let pair2 = Arc::clone(&pair);

std::thread::spawn(move || {
    let (lock, cvar) = &*pair2;
    let mut started = lock.lock().unwrap();
    *started = true;
    cvar.notify_one();
});

let (lock, cvar) = &*pair;
let mut started = lock.lock().unwrap();
while !*started {
    started = cvar.wait(started).unwrap(); // 释放锁并等待，被唤醒后重新获取
}

💡 提示：Condvar::wait 会释放 MutexGuard 并挂起线程；被唤醒后重新获取锁。必须用 while 循环检查条件（防止虚假唤醒）。

🔄 对比：Condvar 的用法与 C++ 的 std::condition_variable、Java 的 Object.wait()/notify() 几乎一致——同样的 while 循环防虚假唤醒模式。

线程本地存储与同步原语

线程本地存储（Thread Local Storage）

use std::cell::RefCell;
thread_local! {
    static COUNTER: RefCell<u32> = RefCell::new(0);
}
COUNTER.with(|c| *c.borrow_mut() += 1);  // each thread has its own copy

💡 提示：TLS 让每个线程拥有独立副本，无需同步即可修改。常用于缓存、随机数生成器、请求上下文等「每线程一份」的状态。因为 thread_local! 中的值不是 Sync 的，所以内部用 RefCell 而非 Mutex。

OnceLock 与 OnceCell（Rust 1.70+）

use std::sync::OnceLock;

static CONFIG: OnceLock<Config> = OnceLock::new();

fn get_config() -> &'static Config {
    CONFIG.get_or_init(|| Config { /* ... */ })  // first call initializes; subsequent calls return cached
}

类型	线程安全	用途
`OnceLock`	是	`static` 声明的全局延迟初始化
`OnceCell`	否	单线程场景的延迟初始化

关键 API：get_or_init（首次访问时初始化）、set（尝试设置，返回 Result）、get_or_try_init（支持返回 Result 的初始化）。

💡 提示：OnceLock 取代了过去的 lazy_static/once_cell 第三方方案，是现在延迟初始化全局变量的标准做法。

Barrier

1
2
3

use std::sync::{Arc, Barrier};
let barrier = Arc::new(Barrier::new(4)); // 4 threads must all arrive
// each thread: barrier.wait(); // blocks until all 4 reach here

🔄 对比：Barrier ≈ C++ 的 std::latch/std::barrier（C++20）、Java 的 CyclicBarrier。

Scope 线程

use std::thread;
let mut numbers = vec![1, 2, 3, 4];
thread::scope(|s| {
    s.spawn(|| println!("{:?}", numbers));      // borrow &numbers
    s.spawn(|| numbers.push(5));                 // borrow &mut numbers
});
// scope exits: all threads joined, numbers usable again
numbers.push(6);

💡 提示：普通 thread::spawn 要求闭包 'static（不能借用局部变量）；thread::scope 借助作用域保证所有线程在作用域结束时必然 join，因此允许借用栈上的局部数据，避免了把数据强行 Arc/move 的繁琐。

⚠️ 注意：scope 内的借用规则仍然生效——编译器会检查同一作用域内的并发借用是否冲突。上例中两个 spawn 的借用不重叠（一个 &T，一个 &mut T），编译器会确保它们不会同时活跃。

异步编程

async/.await 基础

#[tokio::main]
async fn main() {
    let val = async { 42 }.await;
    println!("{}", val);
}

async fn 返回实现 Future 的匿名类型；.await 挂起当前任务，让出线程。
并发原语：join!（并发等待多个 Future）、select!（多路复用）、spawn（任务调度）。

💡 提示：Rust 的异步是「协作式」的——Future 只有被驱动（poll）才会推进，.await 点是让出执行权的地方。运行时（如 tokio）负责调度任务、在 IO 就绪时唤醒它们。这意味着 async fn 本身不执行任何代码，只有 .await 或交给运行时 spawn 后才运行。

🔄 对比：Go 的 goroutine 是抢占式调度（运行时自动切换），Rust 的 async 是协作式（只在 .await 点让出）。好处是切换开销极低且可预测，代价是必须避免在 .await 之间执行长时间阻塞操作。

⚠️ 注意：异步块默认单线程调度，CPU 密集任务应使用 spawn_blocking 或专用线程池，否则会饿死其他任务。

运行时与生态

应用层：使用 tokio 或 async-std 作为运行时。
库层：尽量不直接依赖特定运行时以提高可组合性（使用 futures 兼容 trait）。
IO 密集型服务：HTTP、数据库访问、网络爬虫等使用 async 可获得高并发吞吐。
阻塞操作：避免在 async 任务中直接执行阻塞 IO 或 CPU 密集任务，使用 spawn_blocking 或单独线程池。

⚠️ 注意：在 async 上下文中调用 std::thread::sleep 或 std::net::TcpListener::accept 等同步阻塞操作会阻塞整个执行线程，导致同一线程上的其他任务全部卡住。务必使用运行时提供的异步版本（如 tokio::time::sleep）。

Future trait（进阶）

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct MyFuture { count: u32 }

impl Future for MyFuture {
    type Output = u32;
    fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.count > 0 { self.count -= 1; Poll::Pending }
        else { Poll::Ready(42) }
    }
}

🔬 进阶：Future::poll 是异步的底层机制。返回 Poll::Pending 时，必须通过 cx.waker() 安排唤醒，否则任务将永远不会再被 poll。上例省略了 waker 注册，仅作演示——实际手写 Future 时必须正确处理 waker。

🔬 进阶：Pin 保证自引用 Future 不会被移动，这是 Rust 异步零成本抽象的关键。编译器生成的 Future 状态机可能包含自引用，Pin 在类型层面阻止 mem::swap/mem::replace 等移动操作。

异步 trait

自 Rust 1.75 起，trait 中直接写 async fn 已稳定：

1
2
3

trait AsyncService {
    async fn process(&self, data: &str) -> String;
}

⚠️ 注意：若需要把异步 trait 当作 dyn Trait 动态分派，原生 async fn 在 trait 中返回的是不透明 Future 类型，无法直接 Box<dyn Future>。此时仍需借助 async-trait crate（它把 async fn 转为返回 Pin<Box<dyn Future>>）。

🔬 进阶：async fn in trait 的动态分派限制源于 Rust 的类型系统——每个 async fn 返回的匿名 Future 类型大小不同，无法统一放入 trait object 的虚表。async-trait 通过 Box 擦除类型来绕过，代价是一次堆分配。未来 dyn* 类型可能原生解决此问题。