7Rust程序设计语言

并发编程

fearless concurrency

线程消息传递Send/Sync

🎯 AI 时代的学习建议:不要只读理论。理解这一章的设计动机(为什么需要这个概念?)和边界条件(什么情况下会失效?)。当 AI 生成的代码涉及这些概念时,你能判断它是否用对了场景。AI 能帮你总结内容,但无法替代你亲手阅读、思考、建立直觉的过程。

第七章 并发编程 (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 吗?给出例子。

7/10并发编程
0%