MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Rust中线程的基本概念与用途

2021-09-126.2k 阅读

Rust 中线程的基本概念

在 Rust 编程领域，线程是一个至关重要的概念。线程可以理解为程序执行的一个独立路径，它允许程序同时进行多个任务，极大地提高了程序的执行效率和响应能力。与进程不同，线程共享相同的内存空间，这意味着它们可以更高效地进行数据交互，但同时也带来了数据竞争等挑战。

在操作系统层面，线程是 CPU 调度的基本单位。现代操作系统通常采用分时复用的方式，让多个线程轮流占用 CPU 资源，由于 CPU 切换速度极快，用户感知到的就是多个任务在同时进行。Rust 语言对线程提供了原生支持，通过标准库 std::thread 模块，开发者可以方便地创建和管理线程。

Rust 线程模型

Rust 使用的是 1:1 的线程模型，即一个用户空间线程对应一个内核线程。这种模型的优点是可以充分利用多核 CPU 的优势，每个线程都能真正并行执行。与其他一些语言使用的 M:N 线程模型（多个用户线程映射到多个内核线程）相比，1:1 模型在性能和调度的直接性上具有明显优势。

例如，当一个程序中有多个计算密集型任务时，使用 1:1 线程模型可以将这些任务分配到不同的内核上并行处理，从而大大缩短整个任务的执行时间。

创建简单线程

在 Rust 中创建一个线程非常简单，以下是一个基本示例：

use std::thread;

fn main() {
    let handle = thread::spawn(|| {
        println!("This is a new thread!");
    });

    handle.join().unwrap();
    println!("The new thread has finished.");
}

在上述代码中，thread::spawn 函数用于创建一个新线程。该函数接受一个闭包作为参数，闭包中的代码就是新线程要执行的任务。这里新线程简单地打印了一条消息。

handle.join() 方法用于等待新线程执行完毕。join 方法返回一个 Result，如果线程执行过程中没有发生恐慌（panic），unwrap 方法会返回线程的执行结果（这里我们的线程没有返回值），否则会抛出恐慌。

线程间传递数据

在实际应用中，线程之间往往需要传递数据。Rust 提供了几种机制来实现这一点。

使用 move 闭包传递数据

use std::thread;

fn main() {
    let data = String::from("Hello, thread!");
    let handle = thread::spawn(move || {
        println!("Received data: {}", data);
    });

    handle.join().unwrap();
}

在这个例子中，我们创建了一个字符串 data，然后通过 move 闭包将其所有权转移到新线程中。move 关键字确保闭包获取 data 的所有权，这样新线程就可以使用这个数据。

使用 Arc 和 Mutex 共享数据

Arc（原子引用计数）用于在多个线程间共享数据的所有权，Mutex（互斥锁）则用于保证同一时间只有一个线程可以访问数据，从而避免数据竞争。

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

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

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

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

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

在上述代码中，我们创建了一个 Arc 包裹的 Mutex，里面包含一个初始值为 0 的整数。然后创建了 10 个线程，每个线程获取 Arc 的克隆，并尝试获取 Mutex 的锁来修改共享数据。lock 方法会返回一个 Result，unwrap 方法在这里用于简单地处理可能的错误。最后，主线程等待所有线程完成，并打印出共享数据的最终值。

线程同步机制

由于多个线程共享内存空间，为了避免数据竞争和不一致性，需要使用线程同步机制。

Mutex（互斥锁）

如前面示例所示，Mutex 是 Rust 中最常用的线程同步工具之一。它通过互斥访问来保证同一时间只有一个线程可以访问共享数据。当一个线程获取了 Mutex 的锁，其他线程必须等待锁被释放才能访问数据。

RwLock（读写锁）

RwLock 允许在读取操作远远多于写入操作的场景下提高效率。它区分了读锁和写锁，多个线程可以同时持有读锁来读取数据，但只要有一个线程持有写锁，其他线程就不能持有任何锁，无论是读锁还是写锁。

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

fn main() {
    let shared_data = Arc::new(RwLock::new(String::from("Initial data")));
    let mut handles = vec![];

    for _ in 0..5 {
        let data = Arc::clone(&shared_data);
        let handle = thread::spawn(move || {
            let read_data = data.read().unwrap();
            println!("Read data: {}", read_data);
        });
        handles.push(handle);
    }

    let write_handle = thread::spawn(move || {
        let mut write_data = shared_data.write().unwrap();
        *write_data = String::from("New data");
    });

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

    let final_data = shared_data.read().unwrap();
    println!("Final data: {}", final_data);
}

在这个示例中，我们创建了一个 Arc 包裹的 RwLock，里面包含一个字符串。首先创建了 5 个线程来读取数据，它们可以同时获取读锁。然后创建一个线程来写入数据，写入时会获取写锁，此时其他线程不能获取任何锁。最后，主线程等待所有线程完成，并打印出最终的数据。

Condvar（条件变量）

Condvar 用于线程间的条件同步。它允许一个线程在某个条件满足时通知其他线程。

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

fn main() {
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair_clone = Arc::clone(&pair);

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

    let (lock, cvar) = &*pair;
    let mut started = lock.lock().unwrap();
    while!*started {
        started = cvar.wait(started).unwrap();
    }

    handle.join().unwrap();
    println!("The thread has signaled.");
}

在上述代码中，我们创建了一个 Arc 包裹的 (Mutex, Condvar) 对。新线程获取 Mutex 的锁并修改一个布尔值表示某个条件已满足，然后通过 Condvar 通知一个等待的线程。主线程在条件不满足时，通过 Condvar 的 wait 方法等待通知，wait 方法会释放 Mutex 的锁并阻塞线程，直到收到通知，然后重新获取锁并检查条件。

线程局部存储（TLS）

线程局部存储允许每个线程拥有自己独立的数据副本。在 Rust 中，可以通过 thread_local! 宏来实现。

thread_local! {
    static COUNTER: u32 = 0;
}

fn main() {
    let mut handles = vec![];
    for _ in 0..3 {
        let handle = thread::spawn(|| {
            COUNTER.with(|c| {
                let mut num = *c.borrow_mut();
                num += 1;
                println!("Thread local counter: {}", num);
            });
        });
        handles.push(handle);
    }

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

在这个例子中，我们通过 thread_local! 宏定义了一个线程局部变量 COUNTER。每个线程在访问 COUNTER 时，都会操作自己的副本，互不干扰。通过 with 方法可以对线程局部变量进行读写操作，borrow_mut 方法用于获取可变引用。

线程安全性

Rust 的类型系统和所有权机制为线程安全提供了强大的保障。例如，Send 和 Sync 这两个 trait 在确保线程安全方面起着关键作用。

Send trait

Send trait 标记类型可以安全地在线程间传递所有权。几乎所有 Rust 的基本类型都实现了 Send trait，例如 i32、String 等。如果一个类型的所有成员都实现了 Send trait，那么这个类型也自动实现 Send trait。

use std::thread;

fn main() {
    let data: i32 = 42;
    let handle = thread::spawn(move || {
        println!("Received data: {}", data);
    });

    handle.join().unwrap();
}

在这个例子中，i32 类型实现了 Send trait，所以可以安全地将其所有权转移到新线程中。

Sync trait

Sync trait 标记类型可以安全地在多个线程间共享。同样，大多数基本类型都实现了 Sync trait。如果一个类型的所有成员都实现了 Sync trait，并且该类型的所有方法都是线程安全的，那么这个类型也自动实现 Sync trait。例如，Arc<T> 和 Mutex<T> 当 T: Sync 时，它们本身也实现了 Sync trait，因为它们内部的操作（如引用计数的增减和锁的获取释放）都是线程安全的。

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

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

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

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

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

在这个例子中，Mutex<i32> 实现了 Sync trait，因为 i32 实现了 Sync trait，并且 Mutex 内部的锁操作是线程安全的。所以 Arc<Mutex<i32>> 可以安全地在多个线程间共享。

线程池

在实际应用中，频繁地创建和销毁线程会带来一定的开销。线程池是一种有效的解决方案，它预先创建一定数量的线程，并将任务分配给这些线程执行，避免了重复创建和销毁线程的开销。

使用标准库实现简单线程池

虽然 Rust 标准库没有直接提供线程池的实现，但我们可以基于标准库来构建一个简单的线程池。

use std::sync::{Arc, Mutex};
use std::thread;
use std::collections::VecDeque;
use std::sync::mpsc::{channel, Receiver, Sender};

struct ThreadPool {
    workers: Vec<Worker>,
    sender: Sender<Job>,
}

type Job = Box<dyn FnOnce() + Send + 'static>;

struct Worker {
    id: usize,
    thread: thread::JoinHandle<()>,
}

impl Worker {
    fn new(id: usize, receiver: Receiver<Job>) -> Worker {
        let thread = thread::spawn(move || {
            for job in receiver {
                println!("Worker {} is working on a job.", id);
                job();
            }
            println!("Worker {} is exiting.", id);
        });

        Worker {
            id,
            thread,
        }
    }
}

impl ThreadPool {
    fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        let (sender, receiver) = channel();
        let receiver = Arc::new(Mutex::new(receiver));

        let mut workers = Vec::with_capacity(size);
        for id in 0..size {
            let receiver = Arc::clone(&receiver);
            workers.push(Worker::new(id, receiver));
        }

        ThreadPool {
            workers,
            sender,
        }
    }

    fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let job = Box::new(f);
        self.sender.send(job).unwrap();
    }
}

impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender);
        for worker in &mut self.workers {
            println!("Shutting down worker {}", worker.id);
            if let Err(e) = worker.thread.join() {
                println!("Error joining thread: {}", e);
            }
        }
    }
}

可以这样使用这个线程池：

fn main() {
    let pool = ThreadPool::new(4);

    for i in 0..10 {
        let i = i;
        pool.execute(move || {
            println!("Task {} is running on a worker thread.", i);
        });
    }
}

在上述代码中，ThreadPool 结构体包含一个 Worker 结构体的向量和一个 Sender<Job>。Worker 结构体包含线程的 id 和线程的 JoinHandle。Job 是一个实现了 FnOnce、Send 和 'static 的闭包类型。

ThreadPool::new 方法创建了指定数量的 Worker 线程，并返回一个 ThreadPool 实例。ThreadPool::execute 方法将任务发送到线程池中执行。Drop trait 的实现用于在 ThreadPool 实例销毁时，正确地关闭所有工作线程。

使用第三方库 rayon

rayon 是一个广泛使用的 Rust 线程池库，它提供了更高级和便捷的并行处理功能。

use rayon::prelude::*;

fn main() {
    let data = (0..100).collect::<Vec<_>>();
    let result: i32 = data.par_iter().map(|&x| x * 2).sum();
    println!("Result: {}", result);
}

在这个例子中，我们使用 rayon 的 par_iter 方法将一个普通的迭代器转换为并行迭代器。map 方法对每个元素进行乘以 2 的操作，sum 方法对所有结果进行求和。rayon 会自动管理线程池，将任务分配到多个线程中并行执行，大大提高了计算效率。

线程与异步编程

在 Rust 中，除了线程，异步编程也是实现并发的重要方式。异步编程通过 async/await 语法来处理异步任务，与线程不同，异步任务通常在一个线程中通过事件循环来调度执行。

线程与异步的区别

线程是基于多线程模型，每个线程有自己独立的执行栈，可以真正并行执行任务，适合计算密集型任务。而异步编程基于单线程或少量线程，通过事件循环来调度异步任务，任务在等待 I/O 等操作时会让出执行权，适合 I/O 密集型任务。

例如，一个网络爬虫程序，需要大量的 I/O 操作（如发送 HTTP 请求、读取响应数据），使用异步编程可以在一个线程中高效地处理多个请求，避免了线程切换的开销。而对于一些计算密集型的科学计算任务，使用多线程可以充分利用多核 CPU 的优势，提高计算速度。

结合使用线程与异步

在实际应用中，有时需要结合线程和异步编程来充分发挥两者的优势。例如，可以在一个线程池中使用异步任务来处理 I/O 操作，同时利用线程池的多线程特性来并行处理多个异步任务。

use std::sync::Arc;
use std::thread;
use futures::executor::block_on;
use futures::future::join_all;
use tokio::runtime::Runtime;

async fn async_task(i: i32) -> i32 {
    println!("Async task {} is running.", i);
    // 模拟一些异步操作，如 I/O 等待
    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
    i * 2
}

fn main() {
    let rt = Runtime::new().unwrap();
    let mut handles = vec![];
    for _ in 0..4 {
        let rt = rt.clone();
        let handle = thread::spawn(move || {
            let tasks = (0..10).map(|i| rt.block_on(async_task(i)));
            let result: Vec<i32> = block_on(join_all(tasks)).into_iter().collect();
            println!("Thread result: {:?}", result);
        });
        handles.push(handle);
    }

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

在这个示例中，我们使用 tokio 作为异步运行时，定义了一个异步任务 async_task，它模拟了一个异步 I/O 操作（通过 tokio::time::sleep）。然后在主线程中创建了 4 个线程，每个线程使用 Runtime 来执行多个异步任务，并收集结果。这样既利用了线程的并行性，又利用了异步编程在处理 I/O 方面的高效性。

结语

Rust 中的线程提供了强大而灵活的并发编程能力。通过深入理解线程的基本概念、同步机制、数据传递方式以及与异步编程的结合，开发者可以编写出高效、安全且健壮的并发程序。无论是开发高性能的服务器应用，还是进行大规模数据处理，Rust 的线程模型都能为开发者提供有力的支持。在实际应用中，需要根据具体的任务特性（如计算密集型还是 I/O 密集型）来合理选择线程、异步编程或者两者结合的方式，以达到最佳的性能和资源利用效果。同时，要始终注意线程安全问题，充分利用 Rust 的类型系统和所有权机制来避免数据竞争等并发编程中的常见问题。