Rust 程序设计语言

共享状态并发

消息传递是处理并发的一种好方式，但并不是唯一的方式。 另一种方式是让多个线程访问相同的共享数据。 再回顾一下 Go 语言文档中的那句口号：“不要 通过共享内存来通信。”

通过共享内存来通信会是什么样子？此外，为什么 消息传递的拥护者会告诫不要使用内存共享？

在某种程度上，任何编程语言中的通道都类似于单一所有权， 因为一旦你把值通过通道转移出去，就不应该再使用 那个值。共享内存并发就像多重所有权：多个线程 可以同时访问同一个内存位置。正如你在第 15 章 看到的，智能指针让多重所有权成为可能，多重所有权会 增加复杂性，因为这些不同的所有者需要被管理。Rust 的类型系统 和所有权规则极大地帮助我们正确地完成这种管理。举个 例子，让我们来看看互斥器（mutexes）——一种较常见的用于 共享内存的并发原语。

使用互斥器控制访问

Mutex 是 mutual exclusion（互斥）的缩写，即互斥器 在任意给定时刻只允许一个线程访问某些数据。要访问 互斥器中的数据，线程必须先通过请求获取互斥器的 锁来表明它想要访问。锁（lock）是互斥器的一部分数据结构， 用于跟踪当前谁持有对数据的独占访问权。因此， 互斥器被描述为通过锁系统保护（guarding）它所持有的数据。

互斥器以难以使用而闻名，因为你必须 记住两条规则：

1. 必须在使用数据之前尝试获取锁。
2. 当使用完互斥器所保护的数据时，必须释放锁， 这样其他线程才能获取锁。

要为互斥器找一个现实世界的比喻，想象一个 会议上的小组讨论，只有一个麦克风。在小组成员 发言之前，他们必须询问或示意他们想要使用麦克风。 拿到麦克风后，他们想讲多久就讲多久， 然后把麦克风递给下一个请求发言的小组成员。 如果某个小组成员用完麦克风后忘记将其传递出去， 其他人就都无法发言。如果对共享麦克风的管理 出了问题，讨论就无法按计划进行！

互斥器的管理可能非常棘手，这就是 许多人热衷于使用通道的原因。然而，得益于 Rust 的类型 系统和所有权规则，你无法把锁定和解锁搞错。

Mutex<T> 的 API

作为如何使用互斥器的示例，让我们先在单线程环境中 使用互斥器，如示例 16-12 所示。

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5);

    {
        let mut num = m.lock().unwrap();
        *num = 6;
    }

    println!("m = {m:?}");
}

和许多类型一样，我们使用关联函数 new 来创建 Mutex<T>。 要访问互斥器内部的数据，我们使用 lock 方法来获取 锁。这个调用会阻塞当前线程，使其无法做任何工作， 直到轮到我们持有锁为止。

如果另一个持有锁的线程 panic 了，lock 的调用会失败。在 那种情况下，就没有人能够再获取锁了，所以我们选择 unwrap，让当前线程在处于那种情况时 panic。

获取锁之后，我们可以把返回值——在本例中命名为 num—— 当作内部数据的可变引用来使用。类型系统确保 我们在使用 m 中的值之前先获取锁。m 的类型是 Mutex<i32>，而不是 i32>，所以我们必须调用 lock 才能使用 i32 值。我们不会忘记；类型系统不允许 我们以其他方式访问内部的 i32。

lock 的调用返回一个名为 MutexGuard 的类型，它被包装在 我们已通过 unwrap 处理过的 LockResult 中。MutexGuard 类型 实现了 Deref，指向我们内部的数据；该类型还实现了 Drop， 当 MutexGuard 离开作用域时（即在内部作用域结束时），会自动 释放锁。因此，我们不会有忘记释放锁并 阻塞互斥器被其他线程使用的风险，因为锁的释放是 自动发生的。

在释放锁之后，我们可以打印互斥器的值，并看到我们 成功地把内部的 i32 改成了 6。

对 Mutex<T> 的共享访问

现在让我们尝试使用 Mutex<T> 在多个线程之间共享一个值。我们将 启动 10 个线程，让每个线程把计数器值加 1，这样 计数器就从 0 加到 10。示例 16-13 中的代码会出现编译 错误，我们将利用这个错误来更深入地了解 Mutex<T> 的用法 以及 Rust 如何帮助我们正确地使用它。

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

我们创建了一个 counter 变量来在 Mutex<T> 中持有一个 i32， 就像在示例 16-12 中所做的那样。接下来，我们通过遍历一个数字范围 来创建 10 个线程。我们使用 thread::spawn 并给所有 线程传入相同的闭包：将计数器移动到线程中，通过调用 lock 方法获取 Mutex<T> 上的锁，然后给互斥器中的值加 1。当 线程运行完它的闭包时，num 将离开作用域并释放锁， 从而让另一个线程可以获取它。

在主线程中，我们收集所有的 join 句柄。然后，就像在示例 16-2 中所做的那样，我们对每个句柄调用 join，以确保所有线程 都完成。在那之后，主线程将获取锁并打印这个 程序的结果。

我们提示过这个示例无法编译。现在让我们找出原因！

$ cargo run
   Compiling shared-state v0.1.0 (file:///projects/shared-state)
error[E0382]: borrow of moved value: `counter`
  --> src/main.rs:21:29
   |
 5 |     let counter = Mutex::new(0);
   |         ------- move occurs because `counter` has type `std::sync::Mutex<i32>`, which does not implement the `Copy` trait
...
 8 |     for _ in 0..10 {
   |     -------------- inside of this loop
 9 |         let handle = thread::spawn(move || {
   |                                    ------- value moved into closure here, in previous iteration of loop
...
21 |     println!("Result: {}", *counter.lock().unwrap());
   |                             ^^^^^^^ value borrowed here after move

For more information about this error, try `rustc --explain E0382`.
error: could not compile `shared-state` (bin "shared-state") due to 1 previous error

错误信息表明 counter 值在循环的上一次 迭代中被移动了。Rust 告诉我们，我们无法把锁 counter 的 所有权移动到多个线程中。让我们用在第 15 章讨论过的 多重所有权方法来修复这个编译器错误。

多线程的多重所有权

在第 15 章中，我们通过使用智能指针 Rc<T> 创建一个引用计数的值，让一个值有了多个所有者。让我们在这里做 同样的事情，看看会发生什么。我们将在示例 16-14 中把 Mutex<T> 包装在 Rc<T> 中， 并在把所有权移动到线程之前克隆 Rc<T>。

use std::rc::Rc;
use std::sync::Mutex;
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());
}

再一次，我们编译并得到……不同的错误！编译器教给了我们 很多东西：

$ cargo run
   Compiling shared-state v0.1.0 (file:///projects/shared-state)
error[E0277]: `Rc<std::sync::Mutex<i32>>` cannot be sent between threads safely
  --> src/main.rs:11:36
   |
11 |           let handle = thread::spawn(move || {
   |                        ------------- ^------
   |                        |             |
   |  ______________________|_____________within this `{closure@src/main.rs:11:36: 11:43}`
   | |                      |
   | |                      required by a bound introduced by this call
12 | |             let mut num = counter.lock().unwrap();
13 | |
14 | |             *num += 1;
15 | |         });
   | |_________^ `Rc<std::sync::Mutex<i32>>` cannot be sent between threads safely
   |
   = help: within `{closure@src/main.rs:11:36: 11:43}`, the trait `Send` is not implemented for `Rc<std::sync::Mutex<i32>>`
note: required because it's used within this closure
  --> src/main.rs:11:36
   |
11 |         let handle = thread::spawn(move || {
   |                                    ^^^^^^^
note: required by a bound in `spawn`
  --> /rustc/2d8144b7880597b6e6d3dfd63a9a9efae3f533d3/library/std/src/thread/functions.rs:125:0

For more information about this error, try `rustc --explain E0277`.
error: could not compile `shared-state` (bin "shared-state") due to 1 previous error

哇，这个错误信息很冗长！以下是需要关注的重要部分： `Rc<Mutex<i32>>` cannot be sent between threads safely。编译器 还告诉了我们原因：the trait `Send` is not implemented for `Rc<Mutex<i32>>`。我们将在下一节讨论 Send：它是 确保我们在线程中使用的类型适用于并发场景的 trait 之一。

不幸的是，Rc<T> 在线程之间共享并不安全。当 Rc<T> 管理引用计数时，它在每次调用 clone 时增加计数， 并在每个克隆被丢弃时减少计数。但它没有使用任何 并发原语来确保对计数的更改不会被另一个线程 打断。这可能导致错误的计数——这种细微的 bug 进而可能 导致内存泄漏，或者在我们还没使用完时值就被丢弃了。 我们需要的是一个完全像 Rc<T> 一样的类型，但以线程安全的方式 对引用计数进行更改。

使用 Arc<T> 进行原子引用计数

幸运的是，Arc<T> 正是一种像 Rc<T> 一样可以安全用于 并发场景的类型。a 代表 atomic（原子），意味着它是一种 原子引用计数类型。原子类型是我们在此不详细 介绍的另一种并发原语：请参阅标准库文档中的 std::sync::atomic 了解更多 细节。现在，你只需要知道原子类型像原始 类型一样工作，但可以安全地在线程之间共享。

你可能会想，为什么所有原始类型都不是原子的，以及为什么标准 库类型不默认使用 Arc<T> 来实现。原因是 线程安全会带来性能损失，你只应在真正 需要时才付出代价。如果你只在单线程内对值执行 操作，代码在不需要强制执行原子类型提供的 保证时会运行得更快。

让我们回到示例：Arc<T> 和 Rc<T> 有相同的 API，所以我们通过 修改 use 行、new 调用以及 clone 调用来修复 我们的程序。示例 16-15 中的代码最终可以编译并运行。

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

我们做到了！我们从 0 数到了 10，这看起来可能没什么了不起的， 但它确实教会了我们很多关于 Mutex<T> 和线程安全的知识。 你也可以用这个程序的结构来做比仅仅是递增 计数器更复杂的操作。使用这种策略，你可以把计算分成独立的 部分，将这些部分分配到不同线程上，然后使用 Mutex<T> 让每个线程用各自的部分更新最终结果。

请注意，如果你在做简单的数值运算，标准库中的 std::sync::atomic 模块 提供了 比 Mutex<T> 更简单的类型。这些类型提供了对原始类型 安全、并发、原子的访问。我们在本示例中选择使用带有原始 类型的 Mutex<T>，以便能专注于 Mutex<T> 的工作原理。

比较 RefCell<T>/Rc<T> 与 Mutex<T>/Arc<T>

你可能已经注意到 counter 是不可变的，但我们仍然可以获得 其内部值的可变引用；这意味着 Mutex<T> 提供了 内部可变性，就像 Cell 家族一样。正如我们在第 15 章中使用 RefCell<T> 允许修改 Rc<T> 内部内容一样，我们 使用 Mutex<T> 来修改 Arc<T> 内部的内容。

另一个需要注意的细节是，Rust 无法在使用 Mutex<T> 时保护你 免受各种逻辑错误。回想一下第 15 章，使用 Rc<T> 会有造成引用循环的风险，即两个 Rc<T> 值互相引用， 导致内存泄漏。类似地，Mutex<T> 也有造成 死锁（deadlocks）的风险。当一个操作需要锁定两个资源， 而两个线程各持有一个锁，导致它们 永远互相等待时，就会发生死锁。如果你对死锁感兴趣，可以尝试创建一个 存在死锁的 Rust 程序；然后研究任何语言中针对互斥器的 死锁缓解策略，并尝试在 Rust 中实现它们。 Mutex<T> 和 MutexGuard 的标准库 API 文档提供了 有用的信息。

我们将以讨论 Send 和 Sync trait 以及 如何将它们与自定义类型一起使用来结束本章。