并发与 async

这一课你将能做到

1. 用 std::thread 加 move 闭包跑并行任务，用 mpsc 通道回收结果
2. 用 Arc<Mutex<T>> 在多线程间共享可变状态
3. 理解 async/await 模型，用 tokio 写并发 IO 任务，知道线程与 async 的选型边界

为什么要学

Java 的并发模型（Thread、线程池、synchronized、CompletableFuture）你已熟悉，Rust 的对应物学起来很快，但有一张王牌：数据竞争在 Rust 里是编译错误。「 fearless concurrency」不是口号——Send/Sync 标记加借用检查，把 Java 里要靠 code review 才能发现的竞态提前到编译期。

场景上分两路：判题沙箱是 CPU 密集（多进程并行，std::thread 思路）；网关接口是 IO 密集高并发（async/await，tokio 运行时）。这一课两套都给足代码。

先建立图像

Java                          Rust
new Thread(...)         ↔     std::thread::spawn(|| ...)
ExecutorService         ↔     tokio::spawn（async 任务）rayon（CPU 并行）
BlockingQueue           ↔     std::sync::mpsc 通道
synchronized / Lock     ↔     Mutex<T>（锁与数据绑定）
AtomicInteger           ↔     AtomicU32 / fetch_add
CompletableFuture       ↔     Future + .await
虚拟线程 (Java 21)       ↔     tokio 轻量任务（微秒级切换）

Rust 特有的一条铁律先记下：闭包要用主线程的数据，必须 move 把所有权交出去——「共享」默认被禁止，共享可变性需要 Arc<Mutex> 显式申请。

跟着做

步骤 1：线程与 move 闭包

cargo new conc-demo
cd conc-demo

use std::thread;

fn main() {
    let cases = vec!["case-1".to_string(), "case-2".to_string(), "case-3".to_string()];

    let handles: Vec<_> = cases
        .into_iter()                       // 拿走 vec 所有权，逐个交给线程
        .map(|c| {
            thread::spawn(move || {        // move：把 c 的所有权搬进闭包
                format!("{c} 判定完成")     // 线程内独占使用，无竞态可能
            })
        })
        .collect();

    for h in handles {
        println!("{}", h.join().unwrap()); // join 等待并取回结果
    }
}

去掉 move 编译器直接报错：闭包可能比原变量活得久。这一条「必须 move」的强制，就是 Rust 防数据竞争的第一道闸。

步骤 2：mpsc 通道——多线程提交、单点回收

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

fn main() {
    let (tx, rx) = mpsc::channel();          // 多生产者单消费者

    for id in 1..=3 {
        let tx = tx.clone();                 // 每个线程克隆一份发送端
        thread::spawn(move || {
            let cost = id * 10;              // 模拟判题耗时
            tx.send((id, cost)).unwrap();
        });
    }
    drop(tx);                                // 放弃主发送端，recv 循环才能结束

    let mut passed = 0;
    for (id, cost) in rx {                   // 收满 3 条后通道关闭，循环自然退出
        println!("提交 {id} 判完，耗时 {cost} ms");
        passed += 1;
    }
    println!("共通过 {passed} 个提交");
}

通道传输值的所有权（send 拿走数据），天然无共享无锁。对标 Java 的 BlockingQueue，但消息只能被消费一次，类型上保证。

步骤 3：共享可变状态——Arc<Mutex>

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    let counter = Arc::new(Mutex::new(0u32));   // 原子引用计数 + 互斥锁包裹数据

    let handles: Vec<_> = (0..4)
        .map(|_| {
            let counter = Arc::clone(&counter);  // 克隆的是指针，数据仍一份
            thread::spawn(move || {
                let mut n = counter.lock().unwrap();  // 加锁得到守卫
                *n += 1;                              // 守卫离开作用域自动解锁
            })
        })
        .collect();

    for h in handles { h.join().unwrap(); }
    println!("通过数：{}", *counter.lock().unwrap());
}

和 Java 的本质区别：Java 锁是独立对象（synchronized 块锁谁全凭自觉），Rust 锁与数据在类型上绑定——不拿锁根本访问不到数据，忘了解锁是不存在的（守卫 RAII）。

步骤 4：async/await 与 tokio 入门

先加依赖：

[dependencies]
tokio = { version = "1", features = ["full"] }

use std::time::Duration;

async fn fetch_result(id: u64) -> String {
    tokio::time::sleep(Duration::from_millis(50)).await;   // 模拟查库/调沙箱的 IO 等待
    format!("result-{id}")
}

#[tokio::main]                       // 把 main 变成 async：背后是 tokio 运行时
async fn main() {
    let handles: Vec<_> = (1..=3)
        .map(|id| {
            tokio::spawn(async move {
                fetch_result(id).await
            })
        })
        .collect();

    for h in handles {
        println!("{}", h.await.unwrap());
    }
}

关键认知：async fn 返回的是一个惰性 Future，不 .await 就一行都不执行；tokio::spawn 把任务丢进运行时调度（不是 OS 线程，是绿色任务，单个 runtime 可挂百万级）；.await 让出线程而不是阻塞线程。

步骤 5：并发与并行的边界——一张决策表

// CPU 密集（编译代码、跑用例）→ 真并行 → std::thread / rayon
// IO 密集（查库、调下游 HTTP）→ 高并发 → tokio async
// 混合场景：async 里跑 CPU 密集，必须交给阻塞线程池
use tokio::task;

async fn judge_cpu_bound(code: String) -> String {
    task::spawn_blocking(move || {          // 阻塞任务挪出 async 线程池
        // 模拟重 CPU：真实场景这里调沙箱
        std::thread::sleep(std::time::Duration::from_millis(30));
        format!("done: {code}")
    })
    .await
    .unwrap()
}

#[tokio::main]
async fn main() {
    println!("{}", judge_cpu_bound("print(42)".to_string()).await);
}

在 async 任务里直接 std::thread::sleep 或同步 IO，会占死工作线程拖垮整个 runtime——这是 tokio 迁移者（含 Java 虚拟线程用户）最常踩的坑。

讲清楚要点

1. Send 与 Sync：Send 表示「值可以安全转移到另一个线程」，Sync 表示「&T 可以跨线程共享」。它们是自动推导的标记 trait， Arc<Mutex<T>> 之所以能共享，正是因为它把类型「修补」成了 Send + Sync。
2. 编译期防数据竞争：数据竞争三要素（共享 + 可变 + 无同步）在 Rust 里被 move 语义和借用规则系统性拆掉——要么所有权移交，要么 Arc 加锁。
3. Mutex 是数据的属性：Mutex<T> 包着 T，lock() 返回守卫，解引用才可读写；守卫 drop 即解锁。死锁仍可能（嵌套锁），但「忘了加锁」不可能。
4. async 是状态机不是线程：async fn 编译成状态机，await 点是让出点；tokio 负责调度到少量 OS 线程上。单机十万连接的场景这就是答案。
5. 通道优先于共享内存：Rust 社区箴言「Do not communicate by sharing memory; share memory by communicating」。能通道解决的，不引入 Arc<Mutex>。

你可能会懵的地方

• std::thread 和 tokio 的任务怎么选？ CPU 密集用线程（真并行，核心数即吞吐）；IO 密集高并发用 tokio（阻塞成本为零的等待）。判题执行本身吃 CPU 用线程池，提交网关用 async，两者在真实系统里共存。

• 为什么闭包必须 move？不 move 行不行？ 线程生命周期与主线程无序，闭包借用主线程栈上变量可能变成悬垂引用。Rust 禁止这种「可能危险」，要求 move 转移所有权；跨线程共享请走 Arc。

• Mutex.lock() 返回的为什么是 Result？ 持锁线程 panic 会「毒化」锁，标记数据可能处于不一致状态。unwrap 表示「毒化了就直接崩」；要恢复用 into_inner 或 unwrap_or_else(|e| e.into_inner())。

• async 函数里的普通 Mutex 能用吗？ std Mutex 的守卫不是 Send，跨 await 持有会编译报错（这其实是保护）。要么缩短持锁区间到无 await，要么用 tokio::sync::Mutex（代价更高，仅跨 await 持锁时用）。

• 和 Java 虚拟线程比呢？ 思路同为「阻塞廉价化」：虚拟线程对用户透明（同步代码直接跑），tokio 要求显式 async/await 但换来编译期协作点标记与更精细的调度控制；Rust 侧没有 GC，尾延迟更稳。

错法对照

错法	问题	正解
线程闭包直接借用外部变量	error[E0373] 可能比数据活得久	`move
spawn 后不 join 直接退出	主线程结束，子线程被截断	收集 JoinHandle 逐个 join
for msg in rx 不结束	还有发送端没 drop	用完 drop(tx)，克隆出去的每份都要收尾
async 里调 std::thread::sleep	阻塞工作线程，吞吐崩塌	tokio::time::sleep 或 spawn_blocking
跨 await 持有 std Mutex 守卫	error：守卫非 Send	缩短临界区到 await 之前结束

卡点急救

1. error[E0382] value moved into closure... 多处使用：一个变量 move 给多个线程会冲突。用 Arc::clone 共享，或在循环里 clone 出独立副本。
2. 通道收不到消息/程序提前退出：检查发送端是否全部 drop（含主线程自己的 tx）；send 在接收端被 drop 时报错，通常是接收侧先退出了循环。
3. tokio 程序没输出就结束：Future 是惰性的——#[tokio::main] 只驱动 main 这个 Future；忘 .await 的调用根本没执行。逐个核对 await 与 join。

面试常问

• Rust 如何在编译期保证无数据竞争？ 所有权保证单线程内唯一可变性；Send/Sync 标记约束哪些值能跨线程、哪些引用能共享；Arc 提供受控共享，Mutex 把「访问前必须加锁」做成类型要求。三要素（共享+可变+无同步）在任何组合下都无法同时满足而无编译错误。

• Mutex 与 Java synchronized 的区别？ Rust 的锁与数据类型绑定，无锁则无访问，RAII 自动释放；synchronized 锁对象与受保护数据是弱关联，依赖程序员自觉。另外 Rust 会毒化「持锁时 panic 的锁」，防止脏数据静默传播。

• 线程池 vs async 的选型依据？ 看瓶颈：CPU 密集（编译、压缩、判题执行）选线程并行，任务数≈核心数；IO 密集（网关、聚合调用）选 async，等待期不占线程，单机连接数可上十万。混合系统用 spawn_blocking 桥接，防止阻塞污染 runtime。

自测三问

1. Send 和 Sync 分别约束什么？为什么 Rc<T> 不是 Send 而 Arc<T> 是？
2. mpsc 通道里 rx 的 for 循环什么时候结束？
3. async fn 直接调用不 await 会发生什么？这带来什么坑与好处？

小结一句话

线程拿所有权、通道传所有权、Arc<Mutex> 共享所有权——Rust 把 Java 里最难防的并发 bug 变成了编译错误，async/tokio 则补上高并发 IO 的最后一块拼图。

延伸

• The Rust Book ch16：并发编程
• 下一课：宏与元编程——println! 背后的机制，derive 宏对标 Lombok 的编译期魔法。