16.3 共享状态的并发

16.3.1. 使用共享状态来实现并发

还记得 Go 语言有一句名言是这么说的：

Do not communicate by sharing memory; instead, share memory by communicating. （不要用共享内存来通信，要用通信来共享内存）

上一篇文章 16.2. 使用消息传递来跨线程传递数据就是使用通信的方式来实现并发的。这一篇文章讲一下如何使用共享内存的方式来实现并发。Go 语言不建议使用这种方式，但 Rust 支持通过共享状态来实现并发。

上一篇文章 16.2. 使用消息传递来跨线程传递数据讲的 Channel 类似单所有权：一旦值的所有权转移至 Channel，就无法使用它了。共享内存并发类似于多所有权：多个线程可以同时访问同一块内存。

16.3.2. 使用 `Mutex` 来只允许一个线程来访问数据

Mutex 是 mutual exclusion（互斥锁）的简写。

在同一时刻，Mutex 只允许一个线程来访问某些数据。

想要访问数据，线程必须首先获取互斥锁（lock），在 Rust 里就是调用 lock 方法获得。lock 数据结构是 Mutex 的一部分，它能跟踪谁对数据拥有独占访问权。Mutex 通常被描述为：通过锁定系统来保护它所持有的数据。

16.3.3. `Mutex` 的两条规则

在使用数据之前，必须尝试获取锁（lock）。
使用完 Mutex 所保护的数据，必须对数据进行解锁，以便其他线程可以获取锁。

16.3.4. `Mutex<T>` 的 API

通过 Mutex::new 函数来创建 Mutex<T>，其参数就是要保护的数据。Mutex<T> 实际上是一个智能指针。

在访问数据前，通过 lock 方法来获取锁，这个方法会阻塞当前线程的运行。lock 方法也可能会失败，所以返回的值被 Result 包裹，如果成功，Ok 变体附带的值的类型就为 MutexGuard（智能指针，实现了 Deref 和 Drop）。

看个例子：

use std::sync::Mutex;

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

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

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

使用 Mutex::new 创建了一个互斥锁，其保护的数据是 5，赋给 m。所以 m 的类型是 Mutex<i32>。
后面使用 {} 创建了新的小作用域，在小作用域里使用 lock 方法获取值，使用 unwrap 进行错误处理。由于 MutexGuard 实现了 Deref trait，我们就可以获得内部数据的引用。所以 num 是一个可变引用。
在小作用域内还使用了解引用 * 来修改数据的值为 6。
由于 MutexGuard 实现了 Drop trait，所以在小作用域结束后会自动解锁。
最后打印了修改后的互斥锁内的内容。

输出：

m = Mutex { data: 6, poisoned: false, .. }

16.3.5. 多线程共享 `Mutex<T>`

看个例子：

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 包裹以更好地在多线程中调用，刚开始的值是 0
handles 目前是一个空 Vector
下面通过从 0 到 10（不包括 10）的循环创建了 10 个线程，把每个线程得到的 handle 放到空集合 handles 里。
在线程的闭包里，我们的意图是把 counter 这个互斥锁转移到闭包里（所以使用了 move 关键字），然后获取互斥锁，然后修改它的值，每个线程都加 1。当线程执行完后，num 会离开作用域，互斥锁被释放，其他线程就可以使用了。
从 0 到 10（不包括 10）的循环里还遍历了 handles，使用 join 方法，这样等每个 handle 所对应的线程都结束后才会继续执行。
最后在主线程里尝试获得 counter 的互斥锁，然后把它打印出来。

输出：

$ cargo run
   Compiling shared-state v0.1.0 (/tmp/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 放到多个线程，也就是让多个线程拥有它的所有权呢？

16.3.6. 多线程的多重所有权

在 15.5. Rc<T>：引用计数智能指针与共享所有权讲了一个多重所有权的智能指针叫 Rc<T>，把 counter 用 Rc 包裹即可：

#![allow(unused)]
fn main() {
let counter = Rc::new(Mutex::new(0));
}

在循环里，需要把克隆传进线程，这里用了变量遮蔽把新 counter 值设为旧 counter 的克隆：

#![allow(unused)]
fn main() {
let counter = Rc::clone(&counter);
}

修改后的代码（记得在使用前引入 Rc）：

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

输出：

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`
   --> /Users/stanyin/.rustup/toolchains/stable-aarch64-apple-darwin/lib/rustlib/src/rust/library/std/src/thread/functions.rs:128:8
    |
125 | pub fn spawn<F, T>(f: F) -> JoinHandle<T>
    |        ----- required by a bound in this function
...
128 |     F: Send + 'static,
    |        ^^^^ required by this bound in `spawn`

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<std::sync::Mutex<i32>>` cannot be sent between threads safely，Rc<Mutex<i32>> 不能在线程间安全地传递。编译器也告诉我们了原因：the trait `Send` is not implemented for `Rc<std::sync::Mutex<i32>>`，Rc<Mutex<i32>> 没有实现 Send trait（下一篇文章 16.4. 通过 Send 和 Sync trait 来扩展并发会讲到）。只有实现 Send 的类型才能在线程间安全地传递。

其实在 15.5. Rc<T>：引用计数智能指针与共享所有权讲 Rc<T> 也说到了它不能用于多线程场景：Rc<T> 不能安全地跨线程共享。它不能确保计数的更改不会被另一个线程中断。这可能会导致错误的计数，进而导致内存泄漏或在我们完成之前删除某个值。我们需要的是一种与 Rc<T> 完全相同的类型，但它以线程安全的方式更改引用计数。

那么多线程应该用什么呢？有一个智能指针叫做 Arc<T> 可以胜任这个场景。

16.3.7. 使用 `Arc<T>` 来进行原子引用计数

Arc<T> 和 Rc<T> 类似，但是它可以用于并发场景。Arc 的 A 指的是 Atomic（原子的），这意味着它是一个原子引用计数类型，原子是另一种并发原语。这里不对 Arc<T> 做过于详细的介绍，只需要知道原子像原始类型一样工作，但可以安全地跨线程共享，其余信息详见 Rust 官方文档。

那么为什么所有的基础类型都不是原子的？为什么标准库不默认使用 Arc<T>？因为线程安全会带来性能开销，只有在真正需要时才值得付出。

幸好 Arc<T> 和 Rc<T> 的 API 相同，所以先前的代码就很好改了（记得在使用前引入 Arc）：

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

16.3.8. `RefCell<T>`/`Rc<T>` vs. `Mutex<T>`/`Arc<T>`

Mutex<T> 提供了内部可变性，和 Cell 家族一样。我们一般使用 Rc<T> 包裹 RefCell<T> 以获得一个有内部可变性的共享所有权数据类型。同样的，使用 Mutex<T> 可以改变 Arc<T> 里面的内容。

当使用 Mutex<T> 时，Rust 无法保护你免受各种逻辑错误的影响。使用 Rc<T> 会带来创建引用循环的风险，其中两个 Rc<T> 值相互引用，从而导致内存泄漏。同样，Mutex<T> 也存在产生死锁（deadlock） 的风险。当一个操作需要锁定两个资源并且两个线程各自获取其中一个锁，导致它们永远等待对方时，就会发生这种情况。Mutex<T> 和 MutexGuard 的标准库 API 文档提供了有用的信息。详见：Mutex<T> API 文档和 MutexGuard API 文档。

Keyboard shortcuts

Rust指南