Rust 并发：thread::spawn 与 Arc<Mutex<T>>

Rust 的并发模型以“无畏并发”（fearless concurrency）著称，其核心在于编译器在编译期就能阻止数据竞争。当你需要多个线程共享并修改同一份数据时，thread::spawn 配合 Arc<Mutex<T>> 是最常用且安全的组合。本文将手把手教你如何正确使用它们。

理解基本组件的作用

创建线程：使用 std::thread::spawn 启动一个新线程。它接收一个闭包，并立即返回一个 JoinHandle，用于等待线程结束。

共享所有权：Arc<T>（Atomically Reference Counted）允许多个线程同时拥有同一数据的所有权。它通过原子操作计数引用，确保内存只在最后一个所有者离开时释放。

互斥访问：Mutex<T>（Mutual Exclusion）保证同一时间只有一个线程能访问内部数据。任何尝试获取锁的操作都会阻塞，直到锁可用。

三者结合：Arc<Mutex<T>> 表示“多个线程共享一份被互斥保护的数据”。

实现一个简单的多线程计数器

1. 定义共享数据类型：假设我们要让 10 个线程各自对一个整数加 1，最终结果应为 10。

2. 创建 Arc<Mutex<i32>> 实例：

   use std::sync::{Arc, Mutex};
   use std::thread;
   
   let counter = Arc::new(Mutex::new(0));

3. 准备存储线程句柄的容器：你需要收集所有 JoinHandle，以便后续等待它们完成。

   let mut handles = vec![];

4. 循环启动线程：

   • 克隆 Arc 实例，使得每个线程都持有一份所有权。
   • 移动 克隆后的 Arc 进入闭包（使用 move 关键字）。
   • 锁定 Mutex，获取可变引用。
   • 修改 内部值。

     for _ in 0..10 {
       let counter = Arc::clone(&counter);
       let handle = thread::spawn(move || {
           let mut num = counter.lock().unwrap();
           *num += 1;
       });
       handles.push(handle);
     }

5. 等待所有线程结束：

   for handle in handles {
       handle.join().unwrap();
   }

6. 验证结果：

   println!("Result: {}", *counter.lock().unwrap());

   输出应为 Result: 10。

常见错误与规避方法

错误 1：忘记 move 闭包

如果你写成：

thread::spawn(|| {
    // 使用 counter
});

编译器会报错：counter 不能被 move 到闭包中，因为它来自外部作用域且未显式转移所有权。必须使用 move || 显式将变量移入闭包。

错误 2：未处理 lock() 的 Result

Mutex::lock() 返回 Result<MutexGuard<T>, PoisonError<MutexGuard<T>>>。如果一个线程在持有锁时 panic，mutex 会被“毒化”（poisoned），后续 lock() 调用会返回错误。示例中用 .unwrap() 是为了简化，但在生产代码中应处理错误：

match counter.lock() {
    Ok(mut num) => *num += 1,
    Err(e) => println!("Mutex poisoned: {:?}", e),
}

错误 3：试图在线程间直接传递非 Send 类型

只有实现了 Send trait 的类型才能安全地在线程间传递。Arc<Mutex<T>> 要求 T: Send。例如，Rc<T> 不是 Send，因此不能用于多线程场景。

性能考量与替代方案

虽然 Arc<Mutex<T>> 安全可靠，但频繁加锁可能成为性能瓶颈。根据场景可考虑以下替代：

例如，若只需对计数器做原子增减，可改用 Arc<AtomicUsize>：

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;

let counter = Arc::new(AtomicUsize::new(0));
let mut handles = vec![];

for _ in 0..10 {
    let counter = Arc::clone(&counter);
    let handle = thread::spawn(move || {
        counter.fetch_add(1, Ordering::SeqCst);
    });
    handles.push(handle);
}

for handle in handles {
    handle.join().unwrap();
}

println!("Result: {}", counter.load(Ordering::SeqCst));

调试技巧

打印线程 ID：有助于确认是否真正在多线程运行：

use std::thread;
println!("Thread {:?} incrementing", thread::current().id());

检查锁持有时间：长时间持有锁会导致其他线程阻塞。尽量缩小临界区：

// 好：只在必要时锁住
{
    let mut num = counter.lock().unwrap();
    *num += 1;
} // 锁在此处释放
// 继续做其他不需锁的操作

使用 parking_lot 替代标准库 Mutex：第三方 crate parking_lot 提供更快的 Mutex 和 RwLock，API 几乎一致，只需替换 use 语句。

完整可运行示例

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

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

    for i in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            println!("Thread {} starting", i);
            let mut num = counter.lock().unwrap();
            *num += 1;
            println!("Thread {} done, counter = {}", i, *num);
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Final counter value: {}", *counter.lock().unwrap());
}

编译并运行：

rustc main.rs && ./main

输出顺序可能不同，但最终计数器值一定是 10，且每个线程的日志都会出现。