Rust 的主要目标之一是处理并发编程，保证程序的内存安全和高效运行。通过利用 Rust 的所有权和类型检查系统，使许多并发错误可以在编译时检测到，从而避免了运行时错误。Rust 的 无畏并发（fearless concurrency） 使得编写代码变得更加容易，而且易于重构，不会引入新的错误。
使用线程同时运行代码
Rust 中的线程（threads）是用于同时运行程序中独立部分的功能。在多线程上下文中编程需要仔细思考，因为线程可以同时运行，所以不同线程上代码的运行顺序无法预知，可能会导致问题，例如竞争状态(线程按照不一致的顺序访问数据或资源)、死锁(两个线程彼此等待，阻止两个线程继续执行)或 bug。通过使用 Rust 的线程实现模型，可以减轻使用线程的负面影响。Rust 的标准库使用了一种 1:1 的线程实现模型，即程序使用一个操作系统线程来支持一个语言线程。此外，还有其他线程模型的 crates，这些线程模型做出了与 1:1 模型不同的权衡。
绿色线程是由语言运行时或编程语言本身实现的线程，而系统线程则是由操作系统内核提供的线程。在使用绿色线程时，通常需要将它们映射到系统线程上执行。具体来说，这些绿色线程通常在运行时的线程池中运行，线程池由少量的系统线程负责管理，这些系统线程负责调度和执行绿色线程。
绿色线程模式被称为 M:N 模型，其中 M 个绿色线程对应 N 个操作系统线程，这里 M 和 N 不必相同。
使用 spawn 创建一个线程
创建一个新线程，我们需要调用 thread::spawn 函数并传递一个闭包，包含我们想要在新线程中运行的代码。
use std::thread;
use std::time::Duration;
fn main() {
thread::spawn(|| {
for i in 1..10 {
println!("hi number {} from the spawned thread!", i);
thread::sleep(Duration::from_millis(1));
}
});
for i in 0..5 {
println!("hi number {} from the main thread!", i);
thread::sleep(Duration::from_millis(1));
}
}
当 Rust 程序的主线程完成时，无论它们是否已完成运行，所有生成的线程都会被关闭，故线程并没有打印到9。输出结果类似下图：
扩大主函数循环次数后，运行图：
thread::sleep函数是Rust标准库中的一个函数，用于使当前线程休眠（暂停执行）一段时间，以便让其他线程有机会运行。它接受一个时间参数，以指定休眠的时间长度，可以使用Duration类型或者std::time::Duration模块的方法来指定休眠时间。在多线程编程中，使用thread::sleep函数可以帮助我们控制线程执行的顺序和时间片分配，从而更好地利用系统资源。但需要注意的是，使用thread::sleep并不能保证线程的执行顺序，这取决于操作系统的调度算法和硬件资源的可用情况。thread::sleep函数是阻塞的，不应该在async函数（异步函数）中使用。
join 等待所有线程完成
为防止主线程结束时，子线程未运行完成问题。使用join方法，等待线程完成。
添加join方法后，主函数会在for运行后，等待子线程运行完毕。
移动join代码位置（join调用放在主线程的for循环之前）：
主线程将等待子线程完成，然后运行其for循环，因此输出将不再交错。
在线程中使用 move 闭包
move 关键字，使闭包获得它使用的值的所有权，可通过 move 将值的所有权从一个线程转移到另一个线程。
当子线程中使用主线程数据，但未使用 move 关键字：
use std::thread;
fn main() {
let v = vec![1, 2, 3];
let handle = thread::spawn(|| {
println!("Here's a vector: {:?}", v);
});
handle.join().unwrap();
}
运行图：
错误信息：
error[E0373]: closure may outlive the current function, but it borrows `v`, which is owned by the current function
--> src\main.rs:5:32
|
5 | let handle = thread::spawn(|| {
| ^^ may outlive borrowed value `v`
6 | println!("Here's a vector: {:?}", v);
| - `v` is borrowed here
|
note: function requires argument type to outlive `'static`
--> src\main.rs:5:18
|
5 | let handle = thread::spawn(|| {
| __________________^
6 | | println!("Here's a vector: {:?}", v);
7 | | });
| |______^
help: to force the closure to take ownership of `v` (and any other referenced variables), use the `move` keyword
|
5 | let handle = thread::spawn(move || {
| ++++
For more information about this error, try `rustc --explain E0373`.
error: could not compile `move_thread` due to previous error
因为 Rust 不知道这个新建线程会执行多久，所以无法知晓 v 的引用是否一直有效。例如以下情况：
use std::thread;
fn main() {
let v = vec![1, 2, 3];
let handle = thread::spawn(|| {
println!("Here's a vector: {:?}", v);
});
drop(v); // oh no!
handle.join().unwrap();
}
根据错误提示修改代码：
help: to force the closure to take ownership of `v` (and any other referenced variables), use the `move` keyword | 5 | let handle = thread::spawn(move || {
使用 move 将所有权转移至闭包中：
use std::thread;
fn main() {
let v = vec![1, 2, 3];
let handle = thread::spawn(move || {
println!("Here's a vector: {:?}", v);
});
// drop(v); // error: value used here after move
handle.join().unwrap();
}
运行图：
消息传递（message passing）：在线程中传递信息
通道是一种保证并发安全的消息传递机制，其核心思想是通过消息传递来共享内存，而不是通过共享内存来进行通信。可以把通道看作一个水道，其中有发送器和接收器两个部分。发送器可以向通道中发送数据，而接收器可以从通道中接收数据。通道的一个重要特点是可以保证线程之间的同步，避免了数据竞争的问题。通道不仅可以用于线程间通信，还可以用于不同进程之间的通信。同时，通道的使用也可以避免一些并发编程中常见的陷阱和错误。
use std::sync::mpsc;
fn main() {
let (tx, rx) = mpsc::channel();
// 不可编译，因未做任何事，不知 通道 发送类型
// error: cannot infer type of the type parameter `T` declared on the function `channel`
}
pub fn channel<T>() -> (Sender<T>, Receiver<T>) 创建一个新的异步通道，并返回发送端(tx)和接收端(rx)。通道的实现方式允许一个通道有多个发送端，但只能有一个接收端。当至少有一个发送端存在（包括克隆的发送端），recv操作将阻塞，直到有可用的消息。try_recv不会阻塞，会立即返回Result<T,E>:有消息则返回一个Ok值，没有消息则返回一个Err值。
如果在使用发送端进行发送时接收端已经断开连接，send方法将返回一个SendError。类似地，如果在接收消息时发送端已经断开连接，recv方法将返回一个RecvError。
use std::sync::mpsc;
use std::thread;
use std::sync::mpsc::RecvError;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let val = String::from("hi");
tx.send(val).unwrap()
});
let receive = rx.recv().unwrap();
println!("Got: {}", receive);
assert_eq!(Err(RecvError), rx.recv());
}
通道与所有权转移
并发编程中因所有权的存在会避免很多错误的产生。以下为错误案例（原因为：转移变量所有权后仍使用变量）：
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let val = String::from("hi");
tx.send(val).unwrap();
println!("val is {}", val); // 不可继续使用val
});
let received = rx.recv().unwrap();
println!("Got: {}", received);
}
错误信息：
error[E0382]: borrow of moved value: `val`
--> src\main.rs:10:31
|
8 | let val = String::from("hi");
| --- move occurs because `val` has type `String`, which does not implement the `Copy` trait
9 | tx.send(val).unwrap();
| --- value moved here
10 | println!("val is {}", val);
| ^^^ value borrowed here after move
发送多个值
为证明并发性，使线程发送多个消息，并在每个消息之间暂停一秒钟。以观察到两个线程在通道上进行交互的情况。
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("thread"),
];
for val in vals {
tx.send(val).unwrap();
thread::sleep(Duration::from_secs(1));
}
});
for received in rx {
println!("Got: {}", received);
}
}
主线程中，不再显式调用 recv 函数，将 rx 视为一个迭代器来读取值。因为for循环中没有任何暂停或延迟的代码，可知主线程正等待从线程接收值。
通过克隆创建多个发送端
mpsc是多生产者、单消费者的缩写。
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
let tx1 = tx.clone();
thread::spawn(move || {
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("thread"),
String::from("thread2"),
String::from("thread3"),
];
for val in vals {
tx1.send(val).unwrap();
thread::sleep(Duration::from_secs(1));
}
});
thread::spawn(move || {
let vals = vec![
String::from("more"),
String::from("messages"),
String::from("for"),
String::from("you"),
];
for val in vals{
tx.send(val).unwrap();
thread::sleep(Duration::from_secs(1));
}
});
for received in rx {
println!("Got: {}", received);
}
}
调用clone函数来克隆发送端。获得一个新的发送端，将其传递给第一个线程。同时将原始的发送端传递给第二个生成的线程。这样，我们就创建了两个线程，每个线程都向同一个接收者发送不同的消息。
因为并发性，输出的顺序可能会有所不同，这取决于系统。当在不同的线程中使用不同的thread::sleep值进行时，每次运行的结果将更加不确定。
共享状态并发性
编程语言中的通道（channels）类似于单一所有权（single ownership），因为一旦将值传输到通道中，就无法再使用该值。而共享内存并发性则类似于多重所有权（multiple ownership）：多个线程可以同时访问相同的内存位置。
使用互斥锁实现逐个线程访问数据
互斥锁（Mutex）是互斥（mutual exclusion）的缩写，它在任何给定时间只允许一个线程访问某些数据。要访问互斥锁中的数据，线程必须首先发出请求以获取互斥锁的锁。锁是互斥锁的一部分，它是一个数据结构，用于跟踪当前谁拥有对数据的独占访问权限。因此，互斥锁通过锁定系统，保护其持有的数据。
互斥锁的两个规则：
-
在使用数据之前，必须尝试获取锁。
-
在使用完互斥锁保护的数据后，必须解锁数据，以便其他线程可以获取锁。
Mutex<T> API
单线程环境中使用互斥锁:
use std::sync::Mutex;
fn main() {
let m = Mutex::new(5);
{
let mut num = m.lock().unwrap();
*num = 6;
}
println!("m = {:?}", m);
}
要访问互斥锁内部的数据，需使用lock方法获取锁。这个调用会阻塞当前线程，直到获得锁为止。如果在调用lock时，有另一个线程持有了锁并发生了panic，那么当前线程无法获取到锁。为了处理这种情况，代码中选择使用unwrap函数来解包（unwrap）LockResult类型，如果发生了上述情况，当前线程会发生panic。
因为m的类型是Mutex<i32>，而不是直接的i32类型，因此我们必须调用lock方法来获取对i32值的访问权限。这是为了确保在使用m中的值之前先获取到了锁。如果我们忘记调用lock，类型系统将不允许我们访问内部的i32值。Mutex<T>实际上是一个智能指针。当我们调用lock方法时，它返回一个叫做MutexGuard的智能指针，它包装了被互斥锁保护的数据。MutexGuard实现了Deref trait，使得我们可以通过它访问内部数据。此外，MutexGuard还实现了Drop trait，在MutexGuard超出作用域时自动释放锁，这发生在内部作用域的末尾。这样一来，我们就不会遗忘手动释放锁，从而阻止其他线程使用互斥锁，因为锁的释放是自动进行的。
Mutex<T>在多个线程之间共享一个值
创建10个线程，让它们每个线程将计数器的值增加1，使计数器从0增加到10。（代码无法通过编译）
use std::sync::Mutex;
use std::thread;
fn main() {
let counter = Mutex::new(0);
let mut handles = vec![];
for _ in 0..10 {
let handle = thread::spawn(move || {
let mut num = counter.lock().unwrap();
*num += 1;
});
handles.push(handle);
}
for handle in handles {
handle.join().unwrap();
}
println!("Result: {}", *counter.lock().unwrap());
}
错误信息：
error[E0382]: use of moved value: `counter`
--> src\main.rs:9:36
|
5 | let counter = Mutex::new(0);
| ------- move occurs because `counter` has type `Mutex<i32>`, which does not implement the `Copy` trait
...
9 | let handle = thread::spawn(move || {
| ^^^^^^^ value moved into closure here, in previous iteration of loop
10 | let mut num = counter.lock().unwrap();
| ------- use occurs due to use in closure
For more information about this error, try `rustc --explain E0382`.
因线程是同时运行的，并且它们在获取锁和修改计数器值的时候没有进行同步。这就导致了数据竞争，使得编译器无法保证计数器的值在运行时是正确的。
要解决这个问题，需要在多线程中正确地使用互斥锁。可以通过引入一个Arc（原子引用计数）来解决问题。Arc允许多个线程共享同一个计数器，而不会造成数据竞争。然后，我们在每个线程中使用Arc的克隆来处理计数器。
多线程和多重所有权
通过使用智能指针Rc<T>创建引用计数值来使一个值拥有多个所有者。使用Rc<T>将Mutex<T>进行包装，并在将所有权移交给线程之前克隆Rc<T>
use std::sync::Mutex;
use std::rc::Rc;
use std::thread;
fn main() {
let counter = Rc::new(Mutex::new(0));
let mut handles = vec![];
for _ in 0..10 {
let counter = Rc::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!("Result: {}", *counter.lock().unwrap());
}
错误信息：
error[E0277]: `Rc<Mutex<i32>>` cannot be sent between threads safely
--> src\main.rs:10:36
|
10 | let handle = thread::spawn(move || {
| ------------- ^------
| | |
| ______________________|_____________within this `[closure@src\main.rs:10:36: 10:43]`
| | |
| | required by a bound introduced by this call
11 | | let mut num = counter.lock().unwrap();
12 | | *num += 1;
13 | | });
| |_________^ `Rc<Mutex<i32>>` cannot be sent between threads safely
|
= help: within `[closure@src\main.rs:10:36: 10:43]`, the trait `Send` is not implemented for `Rc<Mutex<i32>>`
关键信息：Rc<Mutex<i32>> cannot be sent between threads safely 即Rc<Mutex<i32>>不能安全地在线程之间发送。Rc<Mutex<i32>>没有实现Send trait。
原子引用计数 Arc
Arc<T>是一种类似于Rc<T>的类型，可以在并发情况下安全使用。其中的Arc代表原子引用计数（Atomic Reference Counting）。有关原子性可查看std::sync::atomic文档。原子引用计数类型的工作方式类似于普通类型，但它可以安全地在线程之间共享。线程安全性会带来性能损失。
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!("Result: {}", *counter.lock().unwrap());
}
// Result: 10
RefCell/Rc 与 Mutex/Arc 的相似性
当我们使用 Mutex<T> 时，我们可以通过获取锁来获得对被保护数据的可变引用，即使 Mutex<T> 本身是不可变的。这就是 Mutex<T> 提供的 内部可变性 特性，类似于 Cell 系列一样。正如第十五章中使用 RefCell 可以改变 Rc 中的内容那样，同样的可以使用 Mutex 来改变 Arc 中的内容。
需要注意的是 Rust 并不能避免所有可能的逻辑错误。就像在第15章使用 Rc<T> 时可能出现引用循环（两个 Rc<T> 值相互引用导致内存泄漏）一样，使用 Mutex<T> 也可能导致死锁。死锁发生在两个或多个线程互相等待对方所拥有的锁，导致它们无法继续执行。这是程序设计中常见的并发问题之一。
可扩展的并发：Sync 和 Send trait
Rust 语言本身的并发功能非常少。到目前为止，在本章中讨论的几乎所有并发特性都是标准库中的，而不是语言本身的一部分。然而，两个并发概念内置在语言中：std::marker 中的 Sync 和 Send 标记trait。
标记 trait（Marker Trait）是一种特殊的 trait，它不包含方法定义，主要用于为类型附加特定的意义或属性，通过静态检查或约束保证类型的安全和正确性。而普通 trait 则定义了具体的行为，通过实现具体的方法来描述类型的行为和功能。
通过 Send 允许线程之间转移所有权
Send 标记trait 表明类型的所有权可以在线程间传递。几乎所有的 Rust 类型都是 Send 的。任何完全由 Send 类型组成的类型也会自动被标记为 Send。几乎所有的基本类型都是 Send，除了裸指针。
但 Rc<T> 不是 Send 类型，因为Rc<T> 被实现为用于单线程场景，这时不需要为拥有线程安全的引用计数而付出性能代价。在多线程下转移 Rc<T> 所有权会得到：the trait Send is not implemented for Rc<Mutex<i32>>。Arc<T>是 Send 的，故可以在并发情况下安全使用。
Sync 允许多线程访问
Sync 标记trait 表明一个实现了 Sync 的类型可以安全的在多个线程中拥有其值的引用。换句话说，任意类型 T 如果 &T（T 的不可变引用）是 Send 的，那么 T 就是 Sync 的，这意味着该引用可以安全地发送到另一个线程。与 Send 类似，原始类型是 Sync 的，完全由 Sync 类型组成的类型也是 Sync。
智能指针 Rc<T>、RefCell<T> 和 Cell<T> 不满足 Sync 特性的要求，因为它们的内部状态更新不是线程安全的。而 Mutex<T> 是 Sync 的，可以在多个线程中安全地共享访问。在多线程编程中，Sync 特性确保了类型可以在多个线程之间安全共享访问，避免了数据竞争和并发问题。
手动实现 Send 和 Sync 是不安全的
由 Send 和 Sync 特性组成的类型也是 Send 和 Sync的，故不必手动实现这些特性。因为他们是标记 trait，甚至都不需要实现任何方法。他们只是用来加强并发相关的不可变性的。
文章评论