【AI开发之Rust】第 12 课：并发模型与同步原语

多线程编程最大的噩梦是数据竞争：两个线程同时读写同一块内存，行为未定义。其他语言的解决方案靠程序员自律 + 各种加锁约定，依然防不胜防；Rust 的方案是把"能不能安全地跨线程共享/传递"变成类型系统的一部分（Send/Sync），在编译期就把绝大多数并发 bug 关进笼子里。

本课先讲线程级并发的两大流派，再把"为什么 Rust 敢说 fearless concurrency"的机制讲透：

流派一：消息传递（channel）—— 通过<span>"传值"</span>同步，无共享即无竞争（Go 风格）
流派二：共享状态（Arc<Mutex<T>>）—— 通过<span>"加锁"</span>同步，锁里锁外都要守规矩
底层保障：<span>Send</span> / <span>Sync</span> 两个 marker <span>trait</span> —— 类型能不能跨线程，编译器替你查

💡 用这张图建立心理模型：Send = 这个值可以"搬家"到别的线程；Sync = 这个值可以同时被多个线程共享引用。绝大多数类型自动满足，少数"不满足的名单"（Rc、RefCell、裸指针）恰好就是你想跨线程用会踩坑的那几个。

12.2 创建线程：thread::spawn + move

<span>use</span> std::thread;
<span>use</span> std::time::Duration;

<span>fn</span> <span>main</span>() {
    <span>// 新线程：闭包在子线程执行</span>
    <span>let</span> <span></span><span>handle</span> = thread::<span>spawn</span>(|| {
        <span>for</span> <span>i</span> <span>in</span> <span>1</span>..=<span>5</span> {
            <span>println!</span>(<span>"子线程 {i}"</span>);
            thread::<span>sleep</span>(Duration::<span>from_millis</span>(<span>50</span>));
        }
    });

    <span>for</span> <span>i</span> <span>in</span> <span>1</span>..=<span>3</span> {
        <span>println!</span>(<span>"主线程 {i}"</span>);
        thread::<span>sleep</span>(Duration::<span>from_millis</span>(<span>50</span>));
    }

    handle.<span>join</span>().<span>unwrap</span>();     <span>// 等待子线程结束（join 返回 Result）</span>
}

关键点：

• spawn 返回 JoinHandle<T>，.join() 阻塞等待并取回子线程闭包的返回值（Result<T, ...>）；
• 主线程结束则整个进程结束——忘了 join 可能看到子线程没跑完；
• 闭包捕获外部变量必须 move（第 9 课讲过 move 闭包），因为子线程可能比 main 活得久，编译器要求它拥有捕获的数据：

<span>use</span> std::thread;

<span>fn</span> <span>main</span>() {
    <span>let</span> <span></span><span>data</span> = <span>vec!</span>[<span>1</span>, <span>2</span>, <span>3</span>];

    <span>let</span> <span></span><span>handle</span> = thread::<span>spawn</span>(<span>move</span> || {      <span>// 必须 move：data 被搬进子线程</span>
        <span>println!</span>(<span>"子线程看到 {:?}"</span>, data);
    });

    <span>// println!("{:?}", data);   // ❌ data 已被 move 进子线程</span>
    handle.<span>join</span>().<span>unwrap</span>();
}

💡 从所有权视角看线程：spawn(move || ...) 就是把一批数据的 owner 转交给新线程；线程结束、闭包被丢弃，数据随之释放——和普通作用域完全同构，只是"作用域"变成了"另一个线程的生命周期"。理解这一点，线程就不再神秘。

12.3 消息传递：std::sync::mpsc（channel）

"生产者—消费者"式同步，值从一端传另一端，传递即移交所有权：

<span>use</span> std::sync::mpsc;
<span>use</span> std::thread;
<span>use</span> std::time::Duration;

<span>fn</span> <span>main</span>() {
    <span>let</span> (tx, rx) = mpsc::<span>channel</span>();        <span>// 发送端 tx，接收端 rx（mpsc: 多生产者单消费者）</span>

    <span>let</span> <span></span><span>tx2</span> = tx.<span>clone</span>();                  <span>// 克隆出发送端，供第二个生产者用</span>

    thread::<span>spawn</span>(<span>move</span> || {                <span>// 生产者 1</span>
        <span>for</span> <span>i</span> <span>in</span> <span>0</span>..<span>3</span> {
            tx.<span>send</span>(<span>format!</span>(<span>"来自线程A: {i}"</span>)).<span>unwrap</span>();
            thread::<span>sleep</span>(Duration::<span>from_millis</span>(<span>30</span>));
        }
    });

    thread::<span>spawn</span>(<span>move</span> || {                <span>// 生产者 2</span>
        <span>for</span> <span>i</span> <span>in</span> <span>0</span>..<span>3</span> {
            tx2.<span>send</span>(<span>format!</span>(<span>"来自线程B: {i}"</span>)).<span>unwrap</span>();
            thread::<span>sleep</span>(Duration::<span>from_millis</span>(<span>30</span>));
        }
    });

    <span>// 消费者：迭代 rx，直到所有发送端都 drop</span>
    <span>for</span> <span>msg</span> <span>in</span> rx {
        <span>println!</span>(<span>"收到: {msg}"</span>);
    }
}

要点：

• send 把值move 进通道，发送后原变量不可用——无竞争的核心：值同一时刻只在一个线程手里；
• rx.iter() / for msg in rx 在所有发送端被 drop 后自动结束（所以发送端要 move 进线程、别留在 main 里挡住结束）；
• recv() 阻塞等待，try_recv() 不阻塞；send 返回 Result（接收端已 drop 会 Err）；
• 设计偏好：能用 channel 表达的并发优先用 channel——它强制"值只在一处"，心智负担最小。

12.4 共享状态：Arc<Mutex> / Arc<RwLock>

12.4.1 为什么是 Arc<Mutex> 组合

线程间共享            <span>→</span> 需要<span>"引用计数 + 可跨线程"</span> <span>→</span> <span>Arc</span><<span>T</span>>（<span>Rc</span> 的多线程版）
共享且要修改         <span>→</span> 需要<span>"互斥锁"</span>              <span>→</span> <span>Mutex</span><<span>T</span>>
组合：<span>Arc</span><<span>Mutex</span><<span>T</span>>>  每个线程持有一个 <span>Arc</span> 句柄，通过锁访问内部 <span>T</span>

<span>use</span> std::sync::{Arc, Mutex};
<span>use</span> std::thread;

<span>fn</span> <span>main</span>() {
    <span>let</span> <span></span><span>counter</span> = Arc::<span>new</span>(Mutex::<span>new</span>(<span>0</span>));
    <span>let</span> <span>mut </span><span>handles</span> = <span>vec!</span>[];

    <span>for</span> <span>_</span> <span>in</span> <span>0</span>..<span>5</span> {
        <span>let</span> <span></span><span>c</span> = Arc::<span>clone</span>(&counter);          <span>// 每个线程一个句柄</span>
        handles.<span>push</span>(thread::<span>spawn</span>(<span>move</span> || {
            <span>for</span> <span>_</span> <span>in</span> <span>0</span>..<span>100</span> {
                <span>let</span> <span>mut </span><span>guard</span> = c.<span>lock</span>().<span>unwrap</span>();   <span>// 拿锁，guard 解引用 = &mut i32</span>
                *guard += <span>1</span>;                         <span>// 临界区内修改</span>
            }                                        <span>// guard 在这里 drop，自动解锁</span>
        }));
    }

    <span>for</span> <span>h</span> <span>in</span> handles {
        h.<span>join</span>().<span>unwrap</span>();
    }
    <span>println!</span>(<span>"最终计数: {}"</span>, *counter.<span>lock</span>().<span>unwrap</span>());   <span>// 500</span>
}

三个必须讲透的细节：

1. lock() 返回 MutexGuard：guard 就是"被锁住的引用"，*guard 等价于 &mut T；guard 离开作用域自动解锁（Drop），不需要手动 unlock；
2. lock() 返回 Result：持锁线程 panic 会导致锁"中毒"（poisoned），后续 lock 返回 Err——.unwrap() 在示例里够用，生产代码要处理（通常把 lock().expect("...") 写清楚，或 .lock().unwrap_or_else(|e| e.into_inner()) 强行拿回锁内容）；
3. Arc::clone 仍只是 +1 引用计数，不拷贝 Mutex 内部数据。

12.4.2 读多写少：RwLock

<span>use</span> std::sync::{Arc, RwLock};
<span>use</span> std::thread;

<span>fn</span> <span>main</span>() {
    <span>let</span> <span></span><span>config</span> = Arc::<span>new</span>(RwLock::<span>new</span>(<span>String</span>::<span>from</span>(<span>"version=1"</span>)));

    <span>// 多个读线程可以同时拿读锁</span>
    <span>let</span> <span>mut </span><span>readers</span> = <span>vec!</span>[];
    <span>for</span> <span>_</span> <span>in</span> <span>0</span>..<span>3</span> {
        <span>let</span> <span></span><span>c</span> = Arc::<span>clone</span>(&config);
        readers.<span>push</span>(thread::<span>spawn</span>(<span>move</span> || {
            <span>let</span> <span></span><span>guard</span> = c.<span>read</span>().<span>unwrap</span>();          <span>// 读锁：可并发</span>
            <span>println!</span>(<span>"读到: {}"</span>, guard.<span>len</span>());
        }));
    }
    <span>// 写线程拿写锁（独占）</span>
    <span>let</span> <span></span><span>c</span> = Arc::<span>clone</span>(&config);
    <span>let</span> <span></span><span>writer</span> = thread::<span>spawn</span>(<span>move</span> || {
        <span>let</span> <span>mut </span><span>guard</span> = c.<span>write</span>().<span>unwrap</span>();          <span>// 写锁：独占</span>
        guard.<span>push_str</span>(<span>"\nversion=2"</span>);
    });

    <span>for</span> <span>r</span> <span>in</span> readers { r.<span>join</span>().<span>unwrap</span>(); }
    writer.<span>join</span>().<span>unwrap</span>();
    <span>println!</span>(<span>"最终: {}"</span>, *config.<span>read</span>().<span>unwrap</span>());
}

⚠️ 两个锁都要注意别在持锁期间调用可能再次拿同一把锁的代码（会死锁或 panic），也别做耗时 IO（把整条线程堵住）。持锁时间 = 最短临界区。

12.5 Send 与 Sync：把并发安全写进类型

12.5.1 两个 marker trait 的定义

<span>// 语义（不是方法，是标记）：</span>
<span>unsafe</span> <span>trait</span> <span>Send</span> {}     <span>// 该类型的值可以安全地移动到另一个线程（转移所有权）</span>
<span>unsafe</span> <span>trait</span> <span>Sync</span> {}     <span>// 该类型的引用可以安全地被多个线程共享（&T 是 Send）</span>

• 几乎所有类型自动实现 Send/Sync：i32、String、Vec、Box、Arc……；
• 例外名单（编译器自动不给它们实现，因为不安全）：

<span>!</span>Send：裸指针 <span>*</span>mut <span>T</span>、Rc<span><</span><span>T</span><span>></span>（计数非原子）、RefCell<span><</span><span>T</span><span>></span> 的某些使用……
<span>!</span>Sync：RefCell<span><</span><span>T</span><span>></span>、Cell<span><</span><span>T</span><span>></span>（内部可变性无锁保护）

12.5.2 它如何"在编译期拦 bug"

<span>use</span> std::rc::Rc;
<span>use</span> std::thread;

<span>fn</span> <span>main</span>() {
    <span>let</span> <span></span><span>rc</span> = Rc::<span>new</span>(<span>5</span>);
    <span>// thread::spawn(move || println!("{rc}"));</span>
    <span>// ❌ error[E0277]: `Rc<i32>` cannot be sent between threads safely</span>
    <span>//    因为 Rc 的引用计数不是原子的，跨线程并发增减会数据竞争。</span>
    <span>//    编译器直接拒绝 spawn —— 你不用等运行期崩。</span>
}

反过来，Arc<T> 通过原子计数实现 Send + Sync，所以能进线程。把你想跨线程的类型往编译报错里一扔，编译器会告诉你它是不是 Send/Sync——这是 Rust 并发安全最优雅的地方。

12.5.3 组合法则

<span>Send</span> + <span>Sync</span> 组合的自动推导：
  T: <span>Send</span> 且 T: <span>Sync</span>    → Arc<T> 可共享可发送（Mutex/RwLock 内置保证）
  自定义 <span>struct</span> 自动 <span>Send</span>/<span>Sync</span>，当且仅当所有字段都 <span>Send</span>/<span>Sync</span>

💡 实战记忆：要跨线程的共享可变状态，配方是 Arc<Mutex<T>> 或 Arc<RwLock<T>>；要跨线程的配置/不可变大对象，Arc<T> 即可；纯消息流用 mpsc channel。12.8 的综合题会把三者串起来。

12.6 读报错专项：并发三兄弟

12.7 心智升级：共享内存 vs 消息传递，怎么选

选消息传递（channel）：
<span>  -</span> 流程天然是"一条流水线"（一个线程算完交给下一个）
<span>  -</span> 想避免锁/死锁的思考
<span>  -</span> 数据只在两线程间来回（一对一）

选共享状态（Arc<span><span><<span>Mutex</span>></span></span>）：
<span>  -</span> 多线程要反复读同一份状态（配置、连接池、统计）
<span>  -</span> 结构本身是"中心的表"，各线程只是改字段
<span>  -</span> 需要随机访问（不是顺序管道）

工程混合也很常见：线程 A 通过 channel 把任务发给线程 B，
B 再把结果写进共享的 Arc<span><RwLock<Vec<...></span>>> 供主线程汇总。

💡 Rust 的 channel 自带所有权转移，天然实现"消息传递即同步"；而 Go 谚语"不要通过共享内存通信，要通过通信共享内存"在 Rust 里同样适用——channel 往往是更不容易错的起点。

12.8 📝 动手练习

参考实现放 code/12-concurrency/（写作时同步给出）。

1. join 与返回值：spawn 一个线程计算 1..=10_000 的和返回，主线程 join 后打印结果。
2. move 捕获：把第 4 课的 String move 例子改造成跨线程版（spawn(move || ...)），记录"不用 move"时的报错并解释。
3. channel 流水线：生产者线程生成 0..100 的平方，消费者求和，经 channel 传值，打印最终和；再增加第二个生产者（用 tx.clone()）。
4. Arc 计数器：复刻 12.4.1，把线程数调到 8、每线程累加 10_000，断言结果 = 80_000（体会无竞争）。再故意改成 Rc，记录 E0277。
5. 读多写少：用 Arc<RwLock<Vec<String>>> 模拟"共享日志"：3 个写线程各 append 若干行，2 个读线程各自打印当前行数，主线程汇总验证行数正确。
6. 综合练习（重点）：做一个"并发单词统计"：主线程把一段长文本按行切成 chunk 交给 channel；N 个工作线程各自统计 chunk 的字频写入 Arc<Mutex<HashMap<String, usize>>>；最后主线程汇总输出 Top3。（提示：任务用 channel 分发时是"分发后 drop 发送端"以结束循环。）
7. Send/Sync 判断表：写出 Rc<T>、Arc<T>、Box<T>、RefCell<T>、Mutex<T>、String 各自 Send/Sync 组合，并解释 Rc/RefCell 为什么不是。

验收门禁：不查资料能解释——channel 为什么天然防数据竞争；Arc 与 Mutex 各自解决什么问题；MutexGuard 的生命周期如何保证自动解锁；Send/Sync 是什么、为什么 Rc/RefCell 不在其中。

✅ 本节小结

• spawn + move：新线程拥有闭包捕获的数据；join 取回结果；
• channel（mpsc）：多生产者单消费者；send 移交所有权 → 值同时只在一处；rx 迭代到发送端全部 drop；
• Arc<Mutex>：原子引用计数共享 + 锁内可变；guard 自动解锁；注意锁中毒与持锁时间；
• Arc<RwLock>：读锁共享、写锁独占，读多写少时用；
• Send/Sync：类型系统的并发安全标记，自动实现；Rc/RefCell/裸指针等例外在编译期拦截；
• 设计决策：流水线用 channel，中心共享状态用 Arc<锁>，两者可混合；
• 记忆配方：跨线程不可变 → Arc<T>；共享可变 → Arc<Mutex/RwLock<T>>；传递数据 → channel。

下一课预告：第 13 课《async/await 异步运行时》——线程阻塞太浪费，Rust 用 async 表达 IO 密集任务：Future 是什么、tokio 如何驱动它、spawn/select!/超时/取消如何编排，以及"异步和线程怎么各司其职"。AI 流式问答的整条 IO 链（第 17-19 课）全靠本课打底。