并发模式与实践
常用并发原语、死锁预防与并发正确性的验证手段。
其他并发原语与模式
Condvar：等待 / 通知
条件变量（condition variable） 用来表达"等某个条件成立再继续"，它必须与 Mutex 配合使用：wait 会原子地释放锁并休眠，被唤醒时重新获取锁。 这就是"检查条件 → 等待"必须用 while 而不是 if 的原因（虚假唤醒 + 条件可能又被别人改掉）。
rust
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::time::Duration;
// 把队列和条件变量打包：它们构成同一个不变式，必须一起使用
struct Buf {
queue: Mutex<VecDeque<u32>>,
not_empty: Condvar, // 通知消费者"有货了"
not_full: Condvar, // 通知生产者"有空位了"
}
fn main() {
let buf = Arc::new(Buf {
queue: Mutex::new(VecDeque::new()),
not_empty: Condvar::new(),
not_full: Condvar::new(),
});
const CAP: usize = 3;
let producer = {
let buf = Arc::clone(&buf);
thread::spawn(move || {
for i in 1..=6 {
let mut q = buf.queue.lock().unwrap();
// while 而不是 if：唤醒后条件可能已被其他线程改变
while q.len() >= CAP {
q = buf.not_full.wait(q).unwrap();
}
q.push_back(i);
println!("生产 {i}");
drop(q); // 先放锁再通知，减少被唤醒线程的空转
buf.not_empty.notify_one();
thread::sleep(Duration::from_millis(5));
}
})
};
let consumer = {
let buf = Arc::clone(&buf);
thread::spawn(move || {
let mut got = Vec::new();
loop {
let mut q = buf.queue.lock().unwrap();
while q.is_empty() {
// wait 原子地释放锁并阻塞；返回时重新持锁
q = buf.not_empty.wait(q).unwrap();
}
let v = q.pop_front().unwrap();
got.push(v);
drop(q);
buf.not_full.notify_one();
if got.len() == 6 { break; }
}
got
})
};
producer.join().unwrap();
println!("消费到 {:?}", consumer.join().unwrap());
}text
输出（生产/消费交错）：
生产 1
生产 2
生产 3
生产 4
生产 5
生产 6
消费到 [1, 2, 3, 4, 5, 6]要点：
wait必须放在while循环里。Condvar允许虚假唤醒（spurious wakeup）；即使没有虚假唤醒，被唤醒后条件也可能已被其他线程消费掉。wait返回LockResult<MutexGuard>，所以要q = buf.cv.wait(q).unwrap()把守卫接回来（当锁中毒时会返回Err）。- 通知有三个方法：
notify_one()（唤醒一个等待者）、notify_all()（全部唤醒，适合"状态变化影响所有等待者"如关机）、wait_while(guard, |q| cond)（1.58+，把 while 循环写得更简洁）。 - 先释放锁再
notify是常见优化（避免被唤醒者立刻又阻塞在锁上），但不是正确性要求。
💡 对照：
Condvar对应 Java 的Object.wait()/notify()或Condition（ReentrantLock.newCondition()）、 C++ 的std::condition_variable、Python 的threading.Condition。用法与陷阱完全一致：永远用 while 包住 wait。
Barrier：让所有线程等到齐
Barrier::new(n) 让前 n-1 个到达的线程阻塞，第 n 个到达时全体放行，常用于"分阶段计算"（每阶段之间需要同步）。
rust
use std::sync::{Arc, Barrier};
use std::thread;
fn main() {
let n = 3;
let barrier = Arc::new(Barrier::new(n));
let handles: Vec<_> = (0..n)
.map(|id| {
let barrier = Arc::clone(&barrier);
thread::spawn(move || {
println!("线程 {id} 完成第一阶段");
barrier.wait(); // 等到三个线程都到齐
println!("线程 {id} 开始第二阶段");
})
})
.collect();
for h in handles { h.join().unwrap(); }
println!("全部完成");
}text
输出：三行"完成第一阶段"一定都在任何"开始第二阶段"之前打印。一次性初始化：OnceLock（1.70+）/ Once
OnceLock<T>（最低稳定版本 1.70）表示"最多被初始化一次"的全局状态，比 Mutex<Option<T>> 更轻、更清晰：
rust
use std::sync::OnceLock;
static CONFIG: OnceLock<String> = OnceLock::new();
fn config() -> &'static String {
// get_or_init 保证并发调用下初始化体只执行一次；后续调用零开销
CONFIG.get_or_init(|| {
println!("（只应看到一次）正在加载配置");
String::from("db=localhost")
})
}
fn main() {
println!("{}", config());
println!("{}", config());
// set 在已初始化时返回 Err（把值还给你）
if let Err(v) = CONFIG.set(String::from("other")) {
println!("已经初始化过了，set 被拒绝: {v}");
}
}text
输出：
（只应看到一次）正在加载配置
db=localhost
db=localhost
已经初始化过了，set 被拒绝: other老式的 std::sync::Once 仍然可用（call_once），但新代码优先用 OnceLock：它把"存值"和"只执行一次"合为一体， 拿结果时不需要再 unwrap 一个 Option。
LazyLock（1.80+）与 thread_local!
LazyLock<T, F>（最低稳定版本 1.80）是"带初始化闭包的 OnceLock"，等价于其他语言的懒加载全局变量：
rust
use std::sync::LazyLock;
// 首次解引用时初始化，之后都是普通静态访问
static NAMES: LazyLock<Vec<String>> =
LazyLock::new(|| vec![String::from("alice"), String::from("bob")]);
fn main() {
// 注意：这里直接当 &Vec<String> 用，语法上不需要 get()
println!("{:?}", *NAMES);
println!("长度 {}", NAMES.len());
}text
输出：
["alice", "bob"]
长度 2Mutex 版本的 LazyLock 也常用于全局可变状态：static CACHE: LazyLock<Mutex<HashMap<..>>>。
线程局部存储（thread-local storage） 用 thread_local! 声明，每个线程一份独立副本，天然免同步：
rust
use std::cell::RefCell;
thread_local! {
// const 初始化（1.59+）可以让编译器做更多优化
static LOCAL_COUNT: RefCell<u32> = const { RefCell::new(0) };
}
fn main() {
LOCAL_COUNT.with(|c| *c.borrow_mut() += 1); // 主线程的副本
std::thread::spawn(|| {
LOCAL_COUNT.with(|c| {
println!("子线程初始值 {}", c.borrow()); // 是 0，不是 1
*c.borrow_mut() += 10;
});
})
.join()
.unwrap();
LOCAL_COUNT.with(|c| println!("主线程仍是 {}", c.borrow()));
}text
输出：
子线程初始值 0
主线程仍是 1⚠️ 陷阱：
thread_local!里的值在线程结束时被销毁。访问另一个已经结束的线程的 TLS 会 panic， 所以 TLS 变量的Drop里不要再触碰别的 TLS 变量（"析构顺序"问题）。另外，thread_local!的访问有轻微开销（读取 TLS 槽位）， 热路径上应把它缓存到局部变量。
工作池：channel + thread::scope
把「多生产者：Sender::clone」与「thread::scope：借用非 'static 数据」组合起来，就是一个最小可用的工作池（worker pool）：不需要 Arc<Mutex<_>>，也不需要 'static。
rust
use std::sync::mpsc;
use std::thread;
fn main() {
let jobs: Vec<u64> = (1..=10).collect();
let results: Vec<(u64, u64)> = thread::scope(|s| {
let (job_tx, job_rx) = mpsc::channel::<u64>();
let (res_tx, res_rx) = mpsc::channel::<(u64, u64)>();
// 共享的 Receiver 用 Arc<Mutex<_>>（也可以改成每 worker 一个通道）
let job_rx = std::sync::Arc::new(std::sync::Mutex::new(job_rx));
for _ in 0..4 {
let job_rx = std::sync::Arc::clone(&job_rx);
let res_tx = res_tx.clone();
s.spawn(move || {
loop {
// 取任务：临界区只有"从通道取一条"，很短
let job = {
let rx = job_rx.lock().unwrap();
match rx.recv() {
Ok(j) => j,
Err(_) => break, // 通道关闭 = 没有更多任务
}
};
res_tx.send((job, job * job)).unwrap();
}
});
}
drop(res_tx); // 主线程不用发结果，必须放弃自己那份
// 主线程投递任务（发生在 scope 内，借用完全合法）
for job in jobs {
job_tx.send(job).unwrap();
}
drop(job_tx); // 投递完毕，worker 的 recv 会返回 Err 从而退出
let mut collected = Vec::new();
while let Ok(r) = res_rx.recv() {
collected.push(r);
}
collected
});
let mut results = results;
results.sort_unstable();
println!("{results:?}");
println!("校验和 = {}", results.iter().map(|(_, sq)| sq).sum::<u64>());
}text
输出：
[(1, 1), (2, 4), (3, 9), (4, 16), (5, 25), (6, 36), (7, 49), (8, 64), (9, 81), (10, 100)]
校验和 = 385🚀 进阶：真实工作池要考虑：任务粒度的均衡（避免长尾）、panic 隔离（一个任务炸了不能拖垮池子）、动态扩缩容、优雅关闭。 生产环境直接用
rayon或tokio的任务池，不要手写。
数据并行：rayon 的 par_iter
如果任务之间无共享可变状态、只做"把集合里的元素各自算一遍再汇总"， 那么连线程和通道都不需要——rayon 提供把普通迭代器换成并行迭代器的能力：
powershell
cargo add rayon
# 得到 Cargo.toml：
# [dependencies]
# rayon = "1"rust
use rayon::prelude::*;
fn main() {
let data: Vec<u64> = (1..=1_000_000).collect();
// par_iter 会自动切分工作、用工作窃取（work-stealing）负载均衡
let total: u64 = data.par_iter().map(|x| x * 2).sum();
println!("total = {total}");
// 也提供了并行排序 / 并行查找等
let mut v = vec![5u32, 3, 9, 1, 7];
v.par_sort_unstable();
println!("{v:?}");
let found = v.par_iter().find_any(|&&x| x > 4);
println!("{found:?}");
}text
预期输出（数值确定）：
total = 1000001000000
[1, 3, 5, 7, 9]
Some(5)⚠️ 说明：
rayon是第三方 crate，本书的编写环境没有网络，因此上面这段代码未经cargo run实测；S计算结果由2 * (1+...+1_000_000) = 1_000_001_000_000手工推导。请在你自己的环境里cargo add rayon后运行验证。🧠 原理：
par_iter之所以能安全地并行，靠的正是 「Send与Sync深入」一节的Send/Sync：rayon 的并行迭代器要求闭包Send + Sync、 元素能被安全共享。所以"编译通过"就已经排除了数据竞争。
数据并行的选择顺序：先用 rayon（集合上的 map/filter/sum/sort）→ 不够灵活再手写 thread::scope + 分块 → 需要 I/O 并发（网络、 文件）则转〈异步编程〉一章 async。
并发正确性实践
数据竞争 vs 竞态条件
| 数据竞争（data race） | 竞态条件（race condition） | |
|---|---|---|
| 定义 | 并发访问同一内存、至少一个写、无同步 | 程序结果依赖线程调度时序 |
| 是否 UB | 是（Rust 里安全代码中不可能出现） | 否，是逻辑 bug |
| 编译器能否发现 | ✅ 安全 Rust 中编译期排除 | ❌ 无法发现 |
| 例子 | 两个线程同时 += 1 同一个 static mut | 先查文件是否存在再创建（TOCTOU） |
下面这个程序能编译、没有数据竞争，但逻辑一定是错的：
rust
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
fn main() {
let value = Arc::new(Mutex::new(0usize));
let handles: Vec<_> = (0..8)
.map(|_| {
let value = Arc::clone(&value);
thread::spawn(move || {
// 读-改-写被拆成两个临界区，中间有足够长的窗口
let cur = *value.lock().unwrap(); // 第一次加锁：只是读
thread::sleep(Duration::from_millis(30));
*value.lock().unwrap() = cur + 1; // 第二次加锁：基于旧值写
})
})
.collect();
for h in handles { h.join().unwrap(); }
println!("value = {}（期望 8）", *value.lock().unwrap());
}text
实际输出：value = 1（期望 8）每次运行都是 1，因为 8 个线程几乎同时读到 cur = 0，然后各自写回 1。Mutex 只保证"每一个临界区内部是原子的"，不保证"两个临界区合起来是原子的"。 修法是把检查与更新合进同一个临界区：
rust
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
fn main() {
let value = Arc::new(Mutex::new(0usize));
let handles: Vec<_> = (0..8)
.map(|_| {
let value = Arc::clone(&value);
thread::spawn(move || {
thread::sleep(Duration::from_millis(30)); // 模拟前置工作
// 读-改-写在一个临界区内完成
let mut guard = value.lock().unwrap();
*guard += 1;
})
})
.collect();
for h in handles { h.join().unwrap(); }
println!("value = {}（期望 8）", *value.lock().unwrap());
}text
输出：value = 8（期望 8）另一个经典竞态是 check-then-act（先检查后执行）："如果缓存里没有就计算并写入"。两个线程可能同时发现"没有"，于是都算一遍。 此时正确的结构是先加锁再检查（或者用 OnceLock / get_or_init 这类原子化的"检查并初始化"）：
rust
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
let cache = Arc::new(Mutex::new(HashMap::<u32, u32>::new()));
let handles: Vec<_> = (0..4)
.map(|_| {
let cache = Arc::clone(&cache);
thread::spawn(move || {
// 错误写法会先 lock 检查、放锁、计算、再 lock 写入 —— 中间窗口导致重复计算
// 正确写法：整个过程一个临界区，检查与写入不会被打断
let mut c = cache.lock().unwrap();
c.entry(7).or_insert_with(|| {
// 真实场景这里是昂贵的计算；entry 保证只执行一次
7 * 7
});
})
})
.collect();
for h in handles { h.join().unwrap(); }
assert_eq!(*cache.lock().unwrap().get(&7).unwrap(), 49);
println!("缓存内容 {:?}", *cache.lock().unwrap());
}text
输出：缓存内容 {7: 49}💡 对照：Java 的
ConcurrentHashMap.computeIfAbsent也是为这个场景设计的； Python 里if key not in d: d[key] = compute()在threading下同样需要显式加锁。
中毒与 unwrap 策略
见「毒化（poisoning）与 unwrap 策略」。补充三条项目级实践建议：
- 不要为了省事到处
unwrap()。lock().unwrap()在"中毒即致命"的服务里可以接受，但要配合panic = "abort"或上层捕获； 库代码更适合把PoisonError转成自己的错误类型并向上传播?。 expect比unwrap好，因为它能写出"为什么这里认为不会失败"：m.lock().expect("配置锁不应中毒")。- 不要在持锁时 panic。把可能 panic 的操作（索引越界、
unwrap、外部调用）移出临界区，是避免中毒最根本的办法。
泄漏锁与跨 await（留给〈异步编程〉一章）
MutexGuard 是 RAII 的，但如果它一直没被 drop，锁就一直被占着。最隐蔽的形态是"守卫的生命周期比你以为的长"：
rust
use std::sync::Mutex;
fn main() {
let m = Mutex::new(vec![1, 2, 3]);
// let len = m.lock().unwrap().len(); // 临时值在语句结束就 drop，锁立刻释放
let guard = m.lock().unwrap(); // 绑定到变量，锁活到作用域结束
println!("len = {}", guard.len());
// 这里如果调用另一个也要 lock() 的函数，就会死锁（不可重入）
drop(guard); // 提前显式释放
println!("锁已释放");
}⚠️ 陷阱（指向〈异步编程〉一章）：在异步代码里，
MutexGuard如果跨越了.await点，任务被挂起期间锁不会释放，会造成严重阻塞甚至死锁， 而且std::sync::MutexGuard不是Send，跨await的Future会直接编译失败。异步环境请用tokio::sync::Mutex， 详见异步编程。
并发测试与 loom
并发 bug 的麻烦在于"不确定复现"。可用的手段：
1) 让测试数据量足够大。 小规模跑不出来，8 线程 × 1000 次 这种量级才有意义。
2) 用 cargo test 的 --test-threads 控制测试并发度。 注意这控制的是测试函数之间的并行度，不是测试内部的线程数：
powershell
cargo test # 默认按 CPU 核数并行跑测试函数
cargo test -- --test-threads=1 # 串行跑测试函数，便于定位共享状态测试的干扰
cargo test -- --nocapture # 显示测试里的 println 输出
cargo test concurrent -- --test-threads=43) 重复运行。 用循环或 cargo test -- --test-threads=1 多次执行；也可以用 --release 跑一遍（优化会改变时序，往往能暴露不同的问题）。
4) 🚀 进阶：loom。 ** loom 是 tokio 团队维护的并发模型检验工具（cargo add loom --dev）， 它把 AtomicUsize、Mutex、Arc 等替换成自己的实现，在所有可能的线程交错**下穷举执行你的代码，从而把"偶发 bug"变成"必然复现"。 代价是只能覆盖很小的状态空间，需要 cfg(loom) 条件编译，通常只为最核心的无锁结构写 loom 测试。用法示意：
rust
// Cargo.toml: [dev-dependencies] loom = "0.7"
// 运行方式：RUSTFLAGS="--cfg loom" cargo test（PowerShell 用 $env:RUSTFLAGS="--cfg loom"）
#[cfg(loom)]
#[test]
fn atomic_counter_is_linearizable() {
loom::model(|| {
// 在 loom 下，Arc / AtomicUsize / thread 都要换成 loom:: 的版本
let c = loom::sync::Arc::new(loom::sync::atomic::AtomicUsize::new(0));
let (c1, c2) = (c.clone(), c.clone());
let seqcst = loom::sync::atomic::Ordering::SeqCst;
let t1 = loom::thread::spawn(move || {
c1.fetch_add(1, seqcst);
});
let t2 = loom::thread::spawn(move || {
c2.fetch_add(1, seqcst);
});
t1.join().unwrap();
t2.join().unwrap();
// loom 会穷举所有合法交错，这个断言任何一个交错失败都会报出来
assert_eq!(c.load(seqcst), 2);
});
}⚠️ 说明：上面这段 loom 测试依赖第三方 crate 且需要
RUSTFLAGS="--cfg loom"，本书环境无网络，未实测；请按 loom 官方文档配置。
什么时候该用异步而不是线程
| 维度 | 线程（〈并发编程〉一章） | 异步（〈异步编程〉一章） |
|---|---|---|
| 任务数量 | 数十到数百 | 数万到数百万 |
| 主要等待 | CPU 计算 | I/O（网络、磁盘、数据库） |
| 单任务栈 | 2 MiB 起（可调） | 状态机大小，通常几十~几百字节 |
| 切换成本 | 微秒级（内核） | 纳秒级（用户态） |
| 表达能力 | 阻塞式代码，直观 | async/await，需运行时（tokio 等） |
| 生态 | std 自带，零依赖 | 需要 tokio/async-std + 异步版库 |
| 并行计算 | ✅ 直接利用多核 | ⚠️ 需 spawn_blocking 或 rayon 配合 |
选择规则：
- CPU 密集（数值计算、图像处理、压缩）→ 用线程或
rayon，目标是吃满所有核。 - I/O 密集（成千上万并发连接）→ 用
async，目标是别让线程白等。 - 两者都有 →
async做编排，把 CPU 密集部分丢进spawn_blocking或 rayon 线程池。 - 任务数只有几十个、逻辑简单 → 就用线程，别上异步（复杂度不值得）。
与其他语言的对照
| 事项 | Java | Python | Go | C++ | Rust |
|---|---|---|---|---|---|
| 创建线程 | new Thread(r).start() | threading.Thread(target=f).start() | go f() | std::thread t(f); | thread::spawn(f) |
| 等待结束 | t.join() | t.join() | wg.Wait()（sync.WaitGroup） | t.join() | h.join() → Result<T, _> |
| 返回值 | 需 Callable + Future | 需 Queue 传回 | channel 接收 | 需 std::future/promise | JoinHandle<T> 直接给 T |
| 互斥 | synchronized / ReentrantLock | threading.Lock（with） | sync.Mutex | std::mutex + lock_guard | Mutex<T> + MutexGuard（RAII） |
| 锁与数据的关系 | 无关（锁代码块） | 无关 | 无关 | 无关 | 锁在类型里（Mutex<T> 内装数据） |
| 读写锁 | ReentrantReadWriteLock | 无内置 | sync.RWMutex | std::shared_mutex | RwLock<T> |
| 原子整数 | AtomicInteger / varhandle | 无内置（GIL 下常不必） | atomic.Int64 | std::atomic<int> | AtomicUsize/AtomicI32/… |
| 内存序 | VarHandle 的 getAcquire/setRelease/compareAndSet | 无对应概念 | atomic 包，默认 SeqCst | std::memory_order 五档 | Ordering 五档（同名同义） |
| 并发哈希表 | ConcurrentHashMap | 无（GIL 保护 dict） | sync.Map | 无标准库版本（需 tbb 等） | 无内置；用 Mutex<HashMap>、DashMap、papaya |
| 消息传递 | BlockingQueue（手动搭） | queue.Queue | channel 是语言级原语 | 无标准库版本 | mpsc::channel（语言无关，库级但一等公民） |
| 线程局部 | ThreadLocal<T> | threading.local() | sync.Map 变通 | thread_local | thread_local! |
| 数据竞争的后果 | 未定义但常"能用" | GIL 使其极难发生 | UB，靠 -race 检测 | UB（C++ 标准明确） | 编译期拒绝（安全代码里不可能） |
| 检测工具 | -Xcomp、JMM 压力测试 | faulthandler、sys.setswitchinterval | go test -race（内置） | TSan/ASan（-fsanitize=thread） | Send/Sync + loom + Miri |
| 取消/超时 | Future.cancel、interrupt | 无安全取消（需 Event） | context.Context | 需手写 | 需手写，或 crossbeam 的 select! + recv_timeout |
| 并发原语数量 | 极多（java.util.concurrent） | 少 | 中等 | 中等 | 少而正交（Mutex/RwLock/Condvar/Barrier/Once/原子） |
| 并行集合操作 | parallelStream() | multiprocessing / concurrent.futures | 无内置 | 无标准库 | rayon 的 par_iter |
几条值得记住的结论：
Arc<Mutex<T>>≈Collections.synchronizedXxx的显式版本，但 Rust 强制你把所有访问路径都走锁， Java 只在你记得加synchronized时才安全。- Go 的 channel 与 Rust 的
mpsc思路最接近，但 Go 的 channel 是多生产者多消费者且可以不关闭（靠 GC 回收）， Rust 的mpsc是单消费者且需要显式drop来关闭；Rust 需要 MPMC 就用crossbeam-channel。 - Python 的 GIL 让"共享可变状态"看起来安全，但只保护单条字节码指令；
x += 1仍然会丢更新。CPU 密集必须用multiprocessing绕开 GIL。 - C++ 的
std::thread与 Rust 最像（都基于 OS 线程、都有原子与内存序、都区分Send/Sync那种"可跨线程性"）， 但 C++ 没有编译期的Send/Sync检查，数据竞争直接是 UB，只能靠 TSan 在运行时抓。
常见坑与编译错误
E0277: Rc cannot be sent between threads safely
报错摘录：
text
error[E0277]: `Rc<RefCell<Vec<i32>>>` cannot be sent between threads safely
= help: within `{closure@src/main.rs:9:19}`, the trait `Send` is not implemented
for `Rc<RefCell<Vec<i32>>>`
note: required by a bound in `spawn`
F: Send + 'static,原因：Rc<T> 的引用计数是普通 usize，两个线程同时 clone/drop 会让计数撕裂，所以标准库用 impl !Send/impl !Sync 明确禁止； RefCell<T> 的运行期借用标志也不是同步的（所以它只是 Send 不 Sync）。
修法：单线程 → 多线程的替换搭档是固定的：
| 单线程 | 多线程 |
|---|---|
Rc<T> | Arc<T> |
RefCell<T> / Cell<T> | Mutex<T> / RwLock<T> / 原子类型 |
Rc<RefCell<T>> | Arc<Mutex<T>> |
E0373: closure may outlive the current function
报错摘录：
text
error[E0373]: closure may outlive the current function, but it borrows `v`,
which is owned by the current function
help: to force the closure to take ownership of `v` ..., use the `move` keyword原因：spawn 要求闭包 'static，而闭包默认按需要借用外部变量。
修法（按需选择）：
rust
use std::thread;
fn main() {
let v = vec![1, 2, 3];
// 修法 1：move —— 交出所有权（主线程不能再使用 v）
let a = v.clone();
let h1 = thread::spawn(move || a.iter().sum::<i32>());
println!("{}", h1.join().unwrap());
// 修法 2：thread::scope —— 借用，不交出所有权
let s = thread::scope(|sc| sc.spawn(|| v.iter().sum::<i32>()).join().unwrap());
println!("{} {:?}", s, v);
}text
输出：
6
6 [1, 2, 3]E0597 / E0521: borrowed data escapes outside of closure / function
报错摘录（把借来的数据搬进 spawn 的闭包并从函数里返回句柄）：
text
error[E0521]: borrowed data escapes outside of function
--> src/main.rs:4:5
|
3 | fn spawn_sum(data: &[i32]) -> thread::JoinHandle<i32> {
| ---- - let's call the lifetime of this reference `'1`
| |
| `data` is a reference that is only valid in the function body
4 | thread::spawn(move || data.iter().sum())
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ `data` escapes the function body here
|
= note: argument requires that `'1` must outlive `'static`如果写的是"在闭包里 &local，然后让引用活过闭包"，则会看到 E0597：
text
error[E0597]: `local` does not live long enough
|
7 | let r = &local;
| ^^^^^^ borrowed value does not live long enough
8 | thread::spawn(...)
| ----------------- argument requires that `local` is borrowed for `'static`原因：move 只搬移它实际用到的变量。如果函数参数本身就是 &[i32]，move 搬移的是"那个引用本身"，而引用指向的数据生命周期仍然短于 'static。
修法（三选一）：
rust
use std::thread;
// 修法 1：把数据的所有权交给函数（调用方 clone）
fn spawn_owned(data: Vec<i32>) -> thread::JoinHandle<i32> {
thread::spawn(move || data.iter().sum())
}
// 修法 2：函数内部用 scope 消化借用，不把句柄传出去
fn sum_in_parallel(data: &[i32]) -> i32 {
let (a, b) = data.split_at(data.len() / 2);
thread::scope(|s| {
let ha = s.spawn(|| a.iter().sum::<i32>());
let hb = s.spawn(|| b.iter().sum::<i32>());
ha.join().unwrap() + hb.join().unwrap()
})
}
// 修法 3：返回 JoinHandle<'static> 只能靠 owned 数据（如上），或改用 rayon::scope
fn main() {
println!("{}", spawn_owned(vec![1, 2, 3]).join().unwrap());
println!("{}", sum_in_parallel(&[1, 2, 3, 4, 5]));
}text
输出：
6
15🧠 原理：想"借用局部数据做并行计算并拿到结果"，正确工具是
thread::scope（作用是把线程的存活期限制在一个词法作用域内），而不是spawn。
Mutex 死锁（编译通过，运行挂死）
现象：程序无输出、CPU 占用接近 0、永不结束。
原因：多个锁的反向获取，或同一把锁被同一线程请求两次（std::sync::Mutex 不可重入）。
rust
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
let a = Arc::new(Mutex::new(1));
let b = Arc::new(Mutex::new(2));
let (a1, b1) = (Arc::clone(&a), Arc::clone(&b));
thread::spawn(move || {
let _x = a1.lock().unwrap();
std::thread::sleep(std::time::Duration::from_millis(50));
let _y = b1.lock().unwrap(); // 等 b，而 b 被主线程拿着
});
let _y = b.lock().unwrap(); // 主线程先拿 b
std::thread::sleep(std::time::Duration::from_millis(50));
let _x = a.lock().unwrap(); // 等 a —— 双向等待，永久阻塞
println!("永远到不了这里");
}修法：
rust
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
fn main() {
let a = Arc::new(Mutex::new(1));
let b = Arc::new(Mutex::new(2));
// 修法：所有线程按同一固定顺序加锁（先用 a 再用 b）
let (a1, b1) = (Arc::clone(&a), Arc::clone(&b));
let t = thread::spawn(move || {
let _x = a1.lock().unwrap();
thread::sleep(Duration::from_millis(10));
let _y = b1.lock().unwrap();
});
{
let _x = a.lock().unwrap();
let _y = b.lock().unwrap();
}
t.join().unwrap();
println!("按固定顺序加锁，无死锁");
// 备选修法：try_lock 拿不到就退让，避免无限等待
let m = Mutex::new(0);
if let Ok(mut g) = m.try_lock() {
*g += 1;
println!("try_lock 成功: {g}");
} else {
println!("锁被占用，稍后重试（生产代码要加退避与超时）");
}
}text
输出：
按固定顺序加锁，无死锁
try_lock 成功: 1Arc<Mutex<T>> 忘记 clone 导致 move 错误
报错摘录：
text
error[E0382]: use of moved value: `counter`
|
6 | let counter = Arc::new(Mutex::new(0));
| ------- move occurs because `counter` has type `Arc<Mutex<i32>>`,
| which does not implement the `Copy` trait
8 | thread::spawn(move || { ... });
| ------- value moved here
12 | println!("{}", counter.lock().unwrap());
| ^^^^^^^ value used here after move
|
= note: consider using `Arc::clone`
help: clone the value to increment its reference count
|
8 | let counter = Arc::clone(&counter);原因：Arc<T> 不是 Copy，move 闭包把它搬走了；循环里第二次迭代更是直接用到了已移走的值。
修法：在 move 之前先 Arc::clone(&x) 出一个局部变量，把局部变量搬进闭包：
rust
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
let counter = Arc::new(Mutex::new(0));
let mut handles = Vec::new();
for _ in 0..4 {
let c = Arc::clone(&counter); // 关键这一行：clone 出局部变量
handles.push(thread::spawn(move || { *c.lock().unwrap() += 1; }));
}
for h in handles { h.join().unwrap(); }
println!("{}", *counter.lock().unwrap()); // 原变量从未被移走
}text
输出：4💡 惯例：写
Arc::clone(&x)而不是x.clone()。前者一眼看出"只增加引用计数"，后者在类型不明确时容易被误读为深拷贝。 Clippy 的clone_on_ref_ptrlint 会强制这个风格。
RwLock 读锁未释放导致写饥饿
现象：写线程长时间（或永远）拿不到锁，读线程却一直很快；吞吐看似正常，但没有进展。
原因（两种独立成因，都会导致同一结果）：
- 读锁持有时间过长：读者在持有读锁的情况下做昂贵计算或阻塞 I/O，写者排队等待。
- 读锁不可重入造成的自我死锁：已持有读锁的线程再次
read()，若此时有写者在排队， 该线程会被自己挡住（WindowsSRWLOCK与 Linuxpthread_rwlock都可能出现）。
rust
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let data = Arc::new(RwLock::new(vec![1u64, 2, 3]));
// 反面示例：读者拿到读锁后做很久的计算（真实代码里常是 I/O 或大循环）
let reader = {
let data = Arc::clone(&data);
thread::spawn(move || {
let r = data.read().unwrap();
let start = Instant::now();
let mut acc = 0u64;
while start.elapsed() < Duration::from_millis(200) {
acc = acc.wrapping_add(r.iter().sum::<u64>()); // 持锁做重活
}
acc
})
};
let writer = {
let data = Arc::clone(&data);
thread::spawn(move || {
let t0 = Instant::now();
data.write().unwrap().push(4); // 必须等读者放锁
println!("写者等待了 {:?}", t0.elapsed());
})
};
reader.join().unwrap();
writer.join().unwrap();
println!("长度 {:?}", data.read().unwrap().len());
}text
输出（写者要等读者把 200ms 的重活干完）：
写者等待了 200ms 左右
长度 4修法：把重活搬出临界区——先在读锁里克隆出快照，立刻放锁，再慢慢算；同时把读操作拆短、避免在持有读锁时调用可能再次加锁的函数。
rust
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::{Duration, Instant};
fn main() {
let data = Arc::new(RwLock::new(vec![1u64, 2, 3]));
let reader = {
let data = Arc::clone(&data);
thread::spawn(move || {
// 只在读锁里做一次廉价的快照拷贝
let snapshot: Vec<u64> = data.read().unwrap().clone();
let start = Instant::now();
let mut acc = 0u64;
while start.elapsed() < Duration::from_millis(200) {
acc = acc.wrapping_add(snapshot.iter().sum::<u64>()); // 无锁计算
}
acc
})
};
let writer = {
let data = Arc::clone(&data);
thread::spawn(move || {
let t0 = Instant::now();
data.write().unwrap().push(4); // 现在几乎立刻拿到写锁
println!("写者等待了 {:?}", t0.elapsed());
})
};
reader.join().unwrap();
writer.join().unwrap();
println!("长度 {:?}", data.read().unwrap().len());
}text
输出：写者等待了 几十微秒 ~ 几毫秒
长度 4速查表
| 需求 | 写法 | 备注 |
|---|---|---|
| 起一个线程并等结果 | let h = thread::spawn(move || 42); h.join().unwrap() | join 返回 Result，panic 会变成 Err |
| 借用局部数据并行 | thread::scope(|s| { s.spawn(|| ..); }) | 1.63+，自动 join |
| 命名/调整栈 | thread::Builder::new().name(n).stack_size(sz).spawn(f) | 返回 io::Result<JoinHandle<T>> |
| 线程 id / 并行度 | thread::current().id() / thread::available_parallelism() | 后者可能返回 Err |
| 单生产者单消费者 | let (tx, rx) = mpsc::channel(); tx.send(v)?; rx.recv()? | 收端全丢时 send 报错 |
| 有界 + 背压 | mpsc::sync_channel(n) | n == 0 为会合通道 |
| 多生产者 | 每个线程 tx.clone()，主线程记得 drop(tx) | 忘记 drop → recv 永久阻塞 |
| 优雅结束接收循环 | while let Ok(m) = rx.recv() { .. } | 不要用 rx.iter()，关闭时会 panic |
| 共享可变状态 | Arc<Mutex<T>> + Arc::clone(&x) | 先 clone 再 move |
| 读多写少 | Arc<RwLock<T>> + read() / write() | 别持读锁做重活 |
| 取锁 | m.lock().unwrap() / m.try_lock() | 中毒时返回 Err(PoisonError) |
| 处理中毒 | Err(e) => e.into_inner() 或 m.clear_poison() | clear_poison 需 1.77+ |
| 无锁计数 | static N: AtomicUsize = AtomicUsize::new(0); N.fetch_add(1, Ordering::Relaxed); | 单变量才够 |
| 发布数据 | 写数据 → flag.store(true, Ordering::Release)；读方 flag.load(Ordering::Acquire) | 成对使用 |
| 等待条件 | while !cond { g = cv.wait(g).unwrap(); } | 必须 while，防虚假唤醒 |
| 全体同步 | Arc<Barrier>::new(n) + barrier.wait() | 分阶段计算 |
| 全局懒初始化 | static X: OnceLock<T> / static X: LazyLock<T> | 1.70+ / 1.80+ |
| 线程局部 | thread_local! { static X: RefCell<u32> = const { .. }; } | 每线程一份 |
| 数据并行 | cargo add rayon + v.par_iter().map(f).sum() | 集合上的 map/filter/sort |
| 并发测试 | cargo test -- --test-threads=1 / --nocapture | 控制测试函数并行度 |