29-Rust 教程 - 并发编程基础

LarryLan

发布于 2026-05-29 13:08:57

210

并发编程基础

多个线程同时跑，编译器：让我先检查一下你的代码有没有"线程安全问题"

🎬 引入

想象一下你在厨房做菜，你一个人切菜、炒菜、洗碗，效率还行但有点慢。于是你叫来了三个朋友帮忙：一个切菜，一个炒菜，一个摆盘。理论上应该快很多，但问题来了：

你们可能会同时去拿同一把刀
炒菜的人可能等切菜的人切完
有人可能把盐当成糖放了

并发编程就是这种情况。多个线程同时工作，理论上能提高效率，但协调不好就会出乱子。

Rust 的并发模型有个特点：编译器会在编译时就帮你检查大部分线程安全问题。这意味着什么？意味着你的代码如果能编译通过，大概率不会有线程竞争、数据竞争这些让人头秃的 bug。

今天我们就来聊聊 Rust 并发编程的基础：如何创建线程、等待线程结束、如何在线程间传递数据，以及最重要的——什么是线程安全。

📌 核心概念

什么是线程？

线程是操作系统能够进行运算调度的最小单位。一个进程可以包含多个线程，它们共享进程的内存空间，但各自有独立的执行栈。

生活化类比：

想象一家餐厅：

进程 = 整个餐厅（有厨房、仓库、收银台）
线程 = 每个服务员（共享餐厅资源，但各自服务不同的客人）

多个服务员可以同时工作，但如果他们都去同一个冰箱拿食材，就需要协调。

Rust 的并发哲学

Rust 的并发模型基于一个核心思想：在编译时消除数据竞争。

数据竞争（Data Race）发生的三个条件：

两个或多个线程同时访问同一数据
至少有一个线程在写入数据
没有使用同步机制

Rust 的所有权系统和类型系统确保：如果你的代码能编译通过，就不会有数据竞争。这不是运行时检查，是编译时保证！

编译器内心 OS： "想同时修改同一个数据？我不允许！除非你用我认可的方式。"

💻 代码示例

创建线程

创建线程最简单的方式是使用 std::thread::spawn 函数：

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

fn main() {
    // 创建一个新线程
    let handle = thread::spawn(|| {
        for i in .. {
            println!("子线程：计数 {}", i);
            thread::sleep(Duration::from_millis());
        }
    });

    // 主线程也做点事
    for i in .. {
        println!("主线程：计数 {}", i);
        thread::sleep(Duration::from_millis());
    }

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

运行结果可能是：

主线程：计数 1
子线程：计数 1
主线程：计数 2
子线程：计数 2
子线程：计数 3
子线程：计数 4
所有线程完成！

注意：输出顺序可能每次都不一样，因为线程调度是操作系统决定的。

join - 等待线程结束

join() 方法会阻塞当前线程，直到目标线程完成。这就像你让朋友去办事，然后你在原地等他回来才能继续下一步。

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

fn main() {
    let handle = thread::spawn(|| {
        println!("子线程开始工作...");
        thread::sleep(Duration::from_secs());
        println!("子线程完成！");
        // 线程可以返回值
    });

    println!("主线程等待中...");
    
    // join 返回线程的返回值
    let result = handle.join().unwrap();
    println!("子线程返回了：{}", result);
}

输出：

主线程等待中...
子线程开始工作...
子线程完成！
子线程返回了：42

move 闭包 - 把数据"搬"进线程

这是新手最容易踩坑的地方。如果你想在线程中使用主线程的数据，需要用 move 关键字：

use std::thread;

fn main() {
    let data = vec![, , , , ];
    
    // ❌ 错误示例 - 这样写编译器会报错
    // let handle = thread::spawn(|| {
    //     println!("数据：{:?}", data);
    // });
    
    // ✅ 正确示例 - 使用 move
    let handle = thread::spawn(move || {
        println!("数据：{:?}", data);
    });
    
    handle.join().unwrap();
    
    // 注意：data 已经移动到子线程了，这里不能再使用
    // println!("{:?}", data);  // ❌ 编译错误！
}

编译器错误信息：

error[E0373]: closure may outlive the current function, but it borrows `data`
  --> src/main.rs:6:32
   |
6  |     let handle = thread::spawn(|| {
   |                                ^^ may outlive borrowed value `data`
   |
note: `data` is borrowed here
  --> src/main.rs:5:9
   |
5  |     let data = vec![1, 2, 3, 4, 5];
   |         ^^^^

编译器在说什么人话？ "你这个闭包可能会活得比 data 还久，但你现在只是借用它。万一主线程把 data 删了，子线程还在用，不就出事了？所以要么你用 move 把数据搬进线程，要么就别用。"

move 的本质： 把数据的所有权"移动"到线程内部，就像你把东西打包快递给朋友，东西就不再是你的了。

线程安全概念

在 Rust 中，线程安全意味着数据可以安全地在多个线程间共享或传递。Rust 通过两个重要的 trait 来标记：

Send - 可以安全地转移到另一个线程
Sync - 可以安全地共享给另一个线程（即 &T 是 Send 的）

大多数类型都是 Send + Sync 的，比如 i32、String、Vec<T> 等。

不是 Send + Sync 的类型：

Rc<T> - 引用计数指针，不是线程安全的
裸指针 *mut T、*const T
某些文件描述符、网络 socket

use std::rc::Rc;
use std::thread;

fn main() {
    let data = Rc::new();
    
    // ❌ 这会编译失败！
    // let handle = thread::spawn(move || {
    //     println!("{}", data);
    // });
    
    // 错误信息：Rc<i32> cannot be sent between threads safely
    // 因为 Rc 不是线程安全的引用计数
}

为什么 Rc 不是线程安全的？ 因为它的引用计数不是原子操作。想象两个线程同时增加引用计数，可能会漏掉一次计数，导致内存提前释放——这就是经典的竞态条件。

🐛 常见坑点

坑点 1：忘记 join 线程

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

fn main() {
    thread::spawn(|| {
        println!("子线程在工作...");
        thread::sleep(Duration::from_secs());
        println!("子线程完成！");
    });
    
    println!("主线程结束");
    // ❌ 没有 join，主线程结束时会直接终止，子线程可能还没跑完
}

可能输出：

主线程结束

（子线程还没来得及输出就被杀掉了）

修复： 保存 JoinHandle 并调用 join()。

坑点 2：在线程中使用循环变量

use std::thread;

fn main() {
    let mut handles = vec![];
    
    for i in .. {
        // ❌ 错误！i 在每次循环中会被复用
        let handle = thread::spawn(|| {
            println!("线程 {}", i);
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

编译器错误：

error[E0373]: closure may outlive the current function, but it borrows `i`

修复方法 1 - 使用 move：

for i in .. {
    let handle = thread::spawn(move || {
        println!("线程 {}", i);
    });
    handles.push(handle);
}

修复方法 2 - 创建新变量：

for i in .. {
    let i = i;  // 创建新的所有权
    let handle = thread::spawn(move || {
        println!("线程 {}", i);
    });
    handles.push(handle);
}

坑点 3：误以为线程按顺序执行

use std::thread;

fn main() {
    let mut data = ;
    
    let h1 = thread::spawn(move || {
        data += ;  // ❌ 编译错误！data 被移动到 h1 了
    });
    
    let h2 = thread::spawn(move || {
        data += ;  // ❌ data 已经被 h1 拿走了
    });
    
    h1.join().unwrap();
    h2.join().unwrap();
    
    println!("{}", data);
}

编译器： "等等，你想让两个线程同时修改同一个变量？这很危险！请用后面要学的 Mutex 或者 channel。"

🎯 实战案例

案例：并行处理大量数据

假设你有一批图片需要处理（比如压缩、滤镜），用单线程会很慢。用多线程可以并行处理：

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

// 模拟处理一张图片
fn process_image(id: usize) {
    println!("开始处理图片 {}", id);
    thread::sleep(Duration::from_millis());  // 模拟耗时操作
    println!("图片 {} 处理完成", id);
}

fn main() {
    let image_ids: Vec<usize> = (..=).collect();
    let mut handles = vec![];
    
    // 创建 4 个工作线程
    let num_workers = ;
    let chunk_size = image_ids.len() / num_workers;
    
    for (i, chunk) in image_ids.chunks(chunk_size).enumerate() {
        let chunk = chunk.to_vec();  // 复制一份数据
        
        let handle = thread::spawn(move || {
            for &id in &chunk {
                process_image(id);
            }
        });
        
        handles.push(handle);
    }
    
    // 等待所有线程完成
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("所有图片处理完成！");
}

输出示例：

开始处理图片 1
开始处理图片 2
开始处理图片 3
开始处理图片 4
图片 1 处理完成
图片 2 处理完成
...
所有图片处理完成！

性能提升： 单线程需要 5 秒，4 线程大约只需要 1.5 秒（因为有调度开销）。

案例：后台任务

有时候你想在后台运行一个任务，不阻塞主程序：

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

fn main() {
    // 启动一个后台保存任务
    let _save_handle = thread::spawn(|| {
        loop {
            println!("自动保存中...");
            thread::sleep(Duration::from_secs());
            // 实际代码中这里会执行保存逻辑
        }
    });
    
    // 主程序继续运行
    for i in ..= {
        println!("主程序工作 {}", i);
        thread::sleep(Duration::from_secs());
    }
    
    // 注意：实际应用中需要优雅地停止后台线程
    // 这里为了简化演示，线程会一直运行
}