第七章并发编程 (Concurrent Programming)

一、导读

1.1 本章学习目标

理解 Rust 的并发编程模型，区分并发（Concurrency）与并行（Parallelism）的概念

掌握使用 std::thread::spawn 创建和管理线程的方法

深入理解 move 闭包在线程间转移所有权的作用

掌握消息传递机制：使用 mpsc channel 在程间安全通信

掌握共享状态并发：使用 Mutex<T> 和 Arc<T> 实现线程间的数据共享

理解 Send 和 Sync 两个标记 trait 的含义及其在并发安全中的作用

了解 Rust 的异步编程模型与线程池的基本概念

能够设计无数据竞争的并发程序，理解"无畏并发"的实现原理

1.2 为什么本章如此重要

并发编程是现代软件开发中不可避免的挑战。随着多核处理器的普及和分布式系统的兴起，编写正确的并发代码变得越来越重要，同时也越来越困难。数据竞争、死锁、活锁等问题在传统的并发编程模型中极难调试——它们可能在测试中从不出现，却在生产环境中偶尔触发，造成难以追踪的 Bug。

Rust 的并发模型是其最引以为傲的特性之一。通过所有权系统和类型系统的配合，Rust 在编译期就能阻止数据竞争的发生。所谓数据竞争（Data Race），必须同时满足三个条件：两个或以上指针同时访问同一数据、至少有一个指针用于写入、没有同步机制。Rust 的借用规则恰好阻止了这种情况——可变引用的独占性确保了写入时不会有其他读者，不可变引用的共享性确保了读取时不会有写入者。

这就是 Rust 所称的"无畏并发"（Fearless Concurrency）——你不需要担心数据竞争，因为编译器会替你检查。这并不意味着 Rust 程序不会死锁或出现逻辑错误，但它确实消除了最危险、最难调试的并发 Bug 类别。

本章将从线程创建开始，逐步讲解消息传递和共享状态两种并发模型，最后介绍 Send 和 Sync trait 如何从类型层面保证并发安全。

1.3 核心问题引导

Rust 如何在编译期阻止数据竞争？它与运行时检测（如 Go 的 race detector）有什么区别？

thread::spawn 的闭包为什么需要 'static 生命周期或 move 关键字？

mpsc channel 的"多生产者单消费者"模型有什么优势？如何实现多消费者？

Mutex<T> 的锁 guard 为什么在离开作用域时自动释放？这与 RAII 有什么关系？

Arc<T> 如何实现所有权的共享？它与引用计数有什么联系？

Send 和 Sync 这两个 trait 分别保证什么？为什么 Rc<T> 不是 Send？

二、核心概念详解

7.1 线程创建与管理

Rust 标准库提供了操作系统原生线程的封装。使用 thread::spawn 创建新线程，它接受一个闭包作为线程的执行体。

rust

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..=5 {
            println!("子线程: {}", i);
            thread::sleep(Duration::from_millis(100));
        }
    });

    for i in 1..=3 {
        println!("主线程: {}", i);
        thread::sleep(Duration::from_millis(150));
    }

    handle.join().unwrap(); // 等待子线程完成
    println!("所有线程已完成");
}

`move` 闭包：默认情况下，闭包会借用环境中的变量。但线程可能比创建它的作用域活得更久，因此闭包必须拥有其使用的数据。move 关键字强制闭包获取环境变量的所有权：

rust

let data = vec![1, 2, 3];

let handle = thread::spawn(move || {
    println!("子线程中的数据: {:?}", data);
});

// data 的所有权已移动到子线程
// println!("{:?}", data); // 编译错误
handle.join().unwrap();

7.2 消息传递：Channel

Channel（通道）是线程间通信的安全方式。Rust 标准库提供了 mpsc（Multiple Producer, Single Consumer）通道。

rust

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

fn main() {
    let (tx, rx) = mpsc::channel();

    // 发送线程
    let handle = thread::spawn(move || {
        let messages = vec!["hello", "from", "thread"];
        for msg in messages {
            tx.send(msg).unwrap();
            thread::sleep(Duration::from_millis(500));
        }
        // tx 在这里被释放，通道关闭
    });

    // 接收线程（主线程）
    for received in rx {
        println!("收到: {}", received);
    }

    handle.join().unwrap();
}

多生产者：通过 clone 发送端可以创建多个生产者：

rust

let (tx, rx) = mpsc::channel();
let tx2 = tx.clone();

thread::spawn(move || {
    tx.send("from thread 1").unwrap();
});

thread::spawn(move || {
    tx2.send("from thread 2").unwrap();
});

同步通道：sync_channel(0) 创建零容量同步通道，发送操作会阻塞直到有接收者就绪，实现线程间的同步。

7.3 共享状态：Mutex 与 Arc

当线程需要共享可变数据时，使用 Mutex<T>（互斥锁）保护数据，使用 Arc<T>（原子引用计数）共享所有权。

rust

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

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

    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);
    }

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

    println!("最终计数: {}", *counter.lock().unwrap());
}

Mutex<T> 的工作原理：Mutex 将数据包装起来，只有通过 lock() 方法获取锁后才能访问数据。锁的 guard（MutexGuard）实现了 Drop trait，在离开作用域时自动释放锁，确保不会忘记解锁。

Arc<T> 的作用：Arc（Atomic Reference Counted）是线程安全版本的 Rc。它使用原子操作维护引用计数，使得多个线程可以安全地共享同一数据的所有权。

7.4 Send 与 Sync Trait

Send 和 Sync 是两个自动实现的标记 trait（marker trait），它们是 Rust 并发安全的基石。

`Send`：表示类型的所有权可以在线程间转移。几乎所有类型都是 Send，除了 Rc<T>（因为引用计数不是原子的）。

`Sync`：表示类型可以安全地被多个线程引用（即 &T 是 Send）。如果 T 是 Sync，那么 &T 可以安全地发送到其他线程。

rust

// 以下代码无法编译，因为 Rc 不是 Send
use std::rc::Rc;
use std::thread;

let data = Rc::new(vec![1, 2, 3]);
thread::spawn(move || {
    println!("{:?}", data); // 编译错误：Rc 不是 Send
});

自动实现规则：

如果类型的所有字段都是 Send，则该类型自动是 Send
如果类型的所有字段都是 Sync，则该类型自动是 Sync
Rc<T> 不是 Send 也不是 Sync
Mutex<T> 是 Send 和 Sync（当 T: Send）
裸指针（*const T、*mut T）既不是 Send 也不是 Sync

7.5 线程安全的集合与同步原语

除了 Mutex 和 Arc，Rust 还提供了其他同步原语：

rust

use std::sync::{RwLock, Condvar, Barrier, Once};

// RwLock：多读单写锁
let lock = RwLock::new(5);
{
    let r = lock.read().unwrap();   // 多个读者可以同时持有
    println!("读取: {}", r);
}
{
    let mut w = lock.write().unwrap(); // 独占写
    *w = 10;
}

// Condvar：条件变量
let pair = Arc::new((Mutex::new(false), Condvar::new()));
let pair2 = Arc::clone(&pair);

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

// Barrier：屏障，等待所有线程到达
let barrier = Arc::new(Barrier::new(3));

三、深入分析与讨论

3.1 消息传递 vs 共享状态

并发编程有两大经典模型：消息传递（"不要通过共享内存来通信，而要通过通信来共享内存"）和共享状态（通过同步原语保护共享数据）。Rust 同时支持两种模型，但鼓励优先使用消息传递。

消息传递的优势在于：数据在任意时刻只有一个所有者，天然避免了数据竞争。Channel 的类型系统还能保证消息的类型安全。共享状态的优势在于：对于需要频繁读写同一数据的场景（如计数器、缓存），避免了消息传递的序列化开销。

在实际项目中，两种模型经常混合使用：使用消息传递在不同模块或组件之间通信，使用共享状态在紧密耦合的并发操作之间共享数据。

3.2 Rust 并发安全的编译期保证

Rust 的并发安全保证可以总结为：如果代码能编译通过，就不会有数据竞争。但这不意味着不会有死锁、活锁或逻辑错误。死锁（两个 Mutex 互相等待对方释放）仍然可能发生，需要开发者通过合理的锁顺序设计来避免。

值得注意的是，Rust 的保证仅限于标准库的 unsafe 边界之内。如果使用 unsafe 代码或第三方库中的 unsafe 实现，可能会绕过编译器的检查。

3.3 线程池与任务调度

在实际应用中，直接创建操作系统线程的开销较大。线程池（Thread Pool）通过复用线程来降低开销。Rust 生态中常用的线程池库包括 rayon（数据并行）和 tokio（异步运行时）。

rust

use rayon::prelude::*;

let sum: u64 = (0..1_000_000).into_par_iter().map(|x| x * x).sum();

四、实践案例

4.1 生产者-消费者模型

使用 channel 实现经典的生产者-消费者并发模式。

rust

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn producer(id: usize, tx: mpsc::Sender<usize>) {
    for i in 0..5 {
        let job = id * 100 + i;
        println!("生产者 {} 发送任务 {}", id, job);
        tx.send(job).unwrap();
        thread::sleep(Duration::from_millis(100));
    }
}

fn consumer(id: usize, rx: mpsc::Receiver<usize>) {
    for job in rx {
        println!("消费者 {} 处理任务 {}", id, job);
        thread::sleep(Duration::from_millis(200));
    }
}

fn main() {
    let (tx, rx) = mpsc::channel();
    
    // 多个生产者
    for i in 0..3 {
        let tx = tx.clone();
        thread::spawn(move || producer(i, tx));
    }
    drop(tx); // 关闭原始发送端
    
    // 单个消费者
    let handle = thread::spawn(move || consumer(0, rx));
    handle.join().unwrap();
    println!("所有任务处理完成");
}

4.2 并发网页下载模拟器

使用 Arc<Mutex<T>> 共享下载状态，模拟并发下载场景。

rust

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

struct DownloadManager {
    completed: Vec<String>,
    total: usize,
}

impl DownloadManager {
    fn new(total: usize) -> Self {
        DownloadManager {
            completed: Vec::new(),
            total,
        }
    }

    fn add_completed(&mut self, url: &str) {
        self.completed.push(url.to_string());
        println!("进度: {}/{}", self.completed.len(), self.total);
    }
}

fn download(url: &str, manager: Arc<Mutex<DownloadManager>>) {
    println!("开始下载: {}", url);
    thread::sleep(Duration::from_millis(500));
    let mut mgr = manager.lock().unwrap();
    mgr.add_completed(url);
}

fn main() {
    let urls = vec![
        "https://example.com/file1.zip",
        "https://example.com/file2.zip",
        "https://example.com/file3.zip",
        "https://example.com/file4.zip",
    ];

    let manager = Arc::new(Mutex::new(DownloadManager::new(urls.len())));
    let mut handles = vec![];

    for url in urls {
        let mgr = Arc::clone(&manager);
        let url = url.to_string();
        handles.push(thread::spawn(move || {
            download(&url, mgr);
        }));
    }

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

    let mgr = manager.lock().unwrap();
    println!("所有下载完成，共 {} 个文件", mgr.completed.len());
}

五、常见误区与注意事项

忘记 `join` 子线程：主线程退出时子线程会被强制终止，即使子线程还没完成任务。必须在主线程中调用 handle.join() 等待子线程完成。

`move` 闭包中的意外所有权转移：使用 move 时，闭包会获取所有引用变量的所有权。如果只需要部分数据，可以先克隆需要的数据再移入闭包。

Mutex 死锁：同一线程两次获取同一个 Mutex 的锁会导致死锁。确保锁的获取顺序一致，避免嵌套锁。

过度使用 `Arc<Mutex<T>>`：如果可以通过消息传递解决问题，优先使用 channel。Arc<Mutex<T>> 引入了锁竞争和潜在的死锁风险。

混淆 `Rc` 和 `Arc`：Rc 不是线程安全的，不能在线程间共享。多线程场景必须使用 Arc。

六、本章小结

本章系统讲解了 Rust 的并发编程模型。Rust 通过所有权系统和类型系统在编译期阻止数据竞争，实现了"无畏并发"。线程通过 thread::spawn 创建，move 闭包确保数据所有权正确转移。消息传递通过 mpsc channel 实现安全的线程间通信；共享状态通过 Mutex<T> 和 Arc<T> 实现受保护的共享访问。Send 和 Sync trait 从类型层面保证了并发安全。Rust 的并发模型在安全性和性能之间取得了优秀的平衡。

七、思考题与练习

编写一个程序，创建 5 个线程，每个线程计算 1 到 1000 中不同范围的素数个数，最后汇总结果。

使用 channel 实现一个"工作窃取"（work-stealing）模型：一个主线程分发任务，多个工作线程处理任务，完成后将结果发送回主线程。

解释为什么 Rc<T> 不是 Send。如果 Rc<T> 在多线程中使用，可能导致什么问题？

使用 Mutex 和 Condvar 实现一个有界缓冲区（bounded buffer），生产者填满时等待，消费者取空时等待。

实现一个线程安全的单例模式。研究 std::sync::Once 的使用方法。

使用 rayon 库将一个大数组的排序改为并行排序，比较性能差异。

编写一个程序，使用 Arc<Mutex<HashMap>> 实现一个简单的线程安全缓存。

解释 Send + Sync 的含义。一个类型可以是 Send 但不是 Sync 吗？给出例子。

第七章 并发编程 (Concurrent Programming)

一、导读

1.1 本章学习目标

1.2 为什么本章如此重要

1.3 核心问题引导

二、核心概念详解

7.1 线程创建与管理

7.2 消息传递：Channel

7.3 共享状态：Mutex 与 Arc

7.4 Send 与 Sync Trait

7.5 线程安全的集合与同步原语

三、深入分析与讨论

3.1 消息传递 vs 共享状态

3.2 Rust 并发安全的编译期保证

3.3 线程池与任务调度

四、实践案例

4.1 生产者-消费者模型

4.2 并发网页下载模拟器

五、常见误区与注意事项

六、本章小结

七、思考题与练习

第七章并发编程 (Concurrent Programming)