❝光阴者，百代之过客，唯有奋力奔跑，方能生风其，是时势造英雄，英雄存在时代❞
大家好，我是「柒八九」。
今天，我们继续「Rust学习笔记」的探索。我们来谈谈关于「Rust学习笔记之并发」的相关知识点。
如果，想了解该系列的文章，可以参考我们已经发布的文章。如下是往期文章。
文章list
- Rust学习笔记之Rust环境配置和入门指南
- Rust学习笔记之基础概念
- Rust学习笔记之所有权
- Rust学习笔记之结构体
- Rust学习笔记之枚举和匹配模式
- Rust学习笔记之包、Crate和模块
- Rust学习笔记之集合
- Rust学习笔记之错误处理
- Rust学习笔记之泛型、trait 与生命周期
- Rust学习笔记之闭包和迭代器
- Rust学习笔记之智能指针
你能所学到的知识点
❝
- 并发编程Concurrent Programming VS 并行编程Parallel programming 「推荐阅读指数」 ⭐️⭐️⭐️
- 使用线程同时运行代码 「推荐阅读指数」 ⭐️⭐️⭐️⭐️⭐️
- 使用消息传递在线程间传送数据 「推荐阅读指数」 ⭐️⭐️⭐️⭐️
- 共享状态并发 「推荐阅读指数」 ⭐️⭐️⭐️
- 使用 Sync 和 Send trait 的可扩展并发 「推荐阅读指数」 ⭐️⭐️⭐️
❞
好了，天不早了，干点正事哇。
并发编程Concurrent programming VS 并行编程Parallel Programming
并发编程Concurrent programming和并行编程Parallel Programming都是指在计算机程序中「同时执行多个任务或操作的编程方式」，但它们在实现方式和目标上存在一些异同点。
❝并发编程Concurrent programming指的是在一个程序中同时进行多个任务，这些任务可以是「独立的，相互之间没有直接的依赖关系」。 ❞
在并发编程中，这些任务通常是「通过交替执行、时间片轮转或事件驱动的方式」来实现并行执行的假象。
并发编程的目标是「提高程序的效率、响应性和资源利用率」。
❝并行编程Parallel Programming是指在「硬件级别上同时执行多个任务，利用计算机系统中的多个处理单元（例如多核处理器）或多台计算机来同时处理多个任务」。 ❞
在并行编程中，任务之间可以「有依赖关系，需要进行任务的分割和协调」。
并行编程的目标是实现「更高的计算性能和吞吐量」。
总结一下，并发编程和并行编程的异同点如下：
- 目标：并发编程旨在提高程序的效率、响应性和资源利用率，而并行编程旨在实现更高的计算性能和吞吐量。
- 执行方式：并发编程通过交替执行、时间片轮转或事件驱动的方式，在一个程序中同时进行多个任务的执行；并行编程通过同时使用多个处理单元或计算机来同时执行多个任务。
- 任务关系：并发编程中的任务通常是独立的，相互之间没有直接的依赖关系；而并行编程中的任务可能存在依赖关系，需要进行任务的分割和协调。
- 实现方式：并发编程可以「通过线程、进程、协程」等机制来实现；并行编程可以通过并行算法、分布式计算等技术来实现。
并发编程Concurrent programming，代表程序的「不同部分相互独立的执行」，而 并行编程parallel programming代表程序「不同部分于同时执行」，这两个概念随着计算机越来越多的利用多处理器的优势时显得愈发重要。
使用线程同时运行代码
在大部分现代操作系统中，已执行程序的代码在一个 进程Process中运行，操作系统则负责管理多个进程。在程序内部，也可以拥有多个同时运行的独立部分。运行这些独立部分的功能被称为 线程Threads。
进程和线程
进程Process和线程Threads是操作系统中用于执行程序的「基本执行单位」，它们之间有着密切的关系，但又有一些本质的区别。
❝进程Process是操作系统中的一个运行实例，它包含了程序执行所需的代码、数据和资源。 ❞
每个进程都有「自己的地址空间，包括独立的堆、栈和全局数据区域」。进程之间是「相互独立的，它们不能直接访问其他进程的内部数据」，通信和数据共享需要通过操作系统提供的机制（如管道、共享内存等）进行。
❝线程Threads是「进程中的一个执行流，它是进程中的一个独立单元，负责执行程序中的指令」。 ❞
「一个进程可以拥有多个线程」，这些线程共享进程的地址空间和资源，包括堆、栈和全局数据区域。不同线程之间可以直接访问进程的内部数据，它们共享相同的上下文环境，因此线程之间的通信和数据共享比进程之间更加高效。
进程和线程之间的关系
- 进程是程序的执行实例，它可以包含多个线程。
- 进程之间是「独立的」，相互之间不能直接访问对方的内部数据，通信需要通过操作系统提供的机制。
- 「同一进程内的多个线程共享进程的地址空间和资源」，它们可以直接访问进程的内部数据。
- 进程之间的切换开销比线程之间的切换开销更大，因为进程切换需要保存和恢复整个进程的上下文环境，而线程切换只需要保存和恢复线程的上下文环境。
- 进程之间的并行执行是由操作系统的调度器决定的，而线程之间的并行执行是由线程调度器（也称为内核级线程调度器或用户级线程调度器）决定的。
线程的问题和类型
将程序中的计算拆分进多个线程可以改善性能，因为程序可以同时进行多个任务，不过这也会增加复杂性。因为线程是同时运行的，所以无法预先保证不同线程中的代码的执行顺序。这会导致诸如此类的问题：
- 竞争状态Race conditions，多个线程以不一致的顺序访问数据或资源
- 死锁Deadlocks，两个线程相互等待对方停止使用其所拥有的资源，这会阻止它们继续运行
- 只会发生在特定情况且难以稳定重现和修复的 bug
1:1线程
很多操作系统提供了「创建新线程的 API」。这种由编程语言调用操作系统 API 创建线程的模型有时被称为 1:1，「一个 OS 线程对应一个语言线程」。
用java创建一个1:1线程
// 创建线程类
class MyThread extends Thread {
public void run() {
// 线程执行的代码
}
}
// 创建并启动线程
public class Main {
public static void main(String[] args) {
MyThread thread = new MyThread();
thread.start();
}
}
绿色线程
很多编程语言提供了自己特殊的线程实现。编程语言提供的线程被称为 绿色green线程，使用绿色线程的语言会「在不同数量的 OS 线程的上下文中执行它们」。为此，绿色线程模式被称为 M:N 模型：M 个绿色线程对应 N 个 OS 线程，这里 M 和 N 不必相同。
用javascript创建一个绿色线程。
// 在 worker.js 文件中的代码
self.onmessage = function(event) {
// 处理消息
// 发送消息回主线程
self.postMessage('处理完成');
};
// 在主线程中的代码
var worker = new Worker('worker.js');
worker.onmessage = function(event) {
// 处理 worker 返回的消息
};
worker.postMessage('开始处理');
Web Worker 引入了一种不同的线程模型，它在 JavaScript 运行时环境中实现了类似线程的机制，但是「并不依赖于操作系统的线程」。Web Worker 在「底层使用了浏览器提供的异步事件模型，利用了浏览器的多线程特性」。
Web Worker 并非真正的操作系统级线程，它是在 JavaScript 运行时环境中模拟的线程。每个 Web Worker 都有自己的上下文和事件循环，它们之间通过消息传递进行通信。因此，Web Worker 通常被称为绿色线程，因为它们在用户级别实现了线程的效果，而不需要依赖操作系统的线程管理。
在当前上下文中，运行时Runtime 代表「二进制文件」中包含的由语言自身提供的代码。这些代码根据语言的不同可大可小，不过「任何非汇编语言都会有一定数量的运行时代码」。为此，通常人们说一个语言 「没有运行时」，一般意味着 「小运行时」。更小的运行时拥有更少的功能不过其优势在于更小的二进制输出，这使其易于在更多上下文中与其他语言相结合。
❝绿色线程的
M:N 模型需要更大的语言运行时来管理这些线程。因此，Rust标准库「只提供了 1:1 线程模型实现」。 ❞
使用 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 1..5 {
println!("hi number {} from the main thread!", i);
thread::sleep(Duration::from_millis(1));
}
}
「当主线程结束时，新线程也会结束，而不管其是否执行完毕」。这个程序的输出可能每次都略有不同。
代码语言：javascript复制hi number 1 from the main thread!
hi number 1 from the spawned thread!
hi number 2 from the main thread!
hi number 2 from the spawned thread!
hi number 3 from the main thread!
hi number 3 from the spawned thread!
hi number 4 from the main thread!
hi number 4 from the spawned thread!
hi number 5 from the spawned thread!
thread::sleep 调用「强制线程停止执行一小段时间」，这会允许其他不同的线程运行。这些线程可能会轮流运行，不过并不保证如此：这「依赖操作系统如何调度线程」。在这里，「主线程首先打印，即便新创建线程的打印语句位于程序的开头」，甚至即便我们告诉新建的线程打印直到 i 等于 9 ，它在主线程结束之前也只打印到了 5。
使用 join 等待所有线程结束
由于主线程结束，上面的代码大部分时候不光会提早结束新建线程，甚至不能实际保证新建线程会被执行。其原因在于无法保证线程运行的顺序！
可以通过将 thread::spawn 的「返回值储存在变量中来修复新建线程部分没有执行或者完全没有执行的问题」。thread::spawn 的返回值类型是 JoinHandle。JoinHandle 是一个「拥有所有权的值」，当「对其调用 join 方法时，它会等待其线程结束」。
use std::thread;
use std::time::Duration;
fn main() {
let handle = thread::spawn(|| {
for i in 1..10 {
println!("hi number {} from the spawned thread!", i);
thread::sleep(Duration::from_millis(1));
}
});
for i in 1..5 {
println!("hi number {} from the main thread!", i);
thread::sleep(Duration::from_millis(1));
}
handle.join().unwrap();
}
通过调用 handle 的 join 会阻塞当前线程直到 handle 所代表的线程结束。阻塞Blocking线程意味着阻止该线程执行工作或退出。
运行上面的代码应该会产生类似这样的输出：
代码语言：javascript复制hi number 1 from the main thread!
hi number 2 from the main thread!
hi number 1 from the spawned thread!
hi number 3 from the main thread!
hi number 2 from the spawned thread!
hi number 4 from the main thread!
hi number 3 from the spawned thread!
hi number 4 from the spawned thread!
hi number 5 from the spawned thread!
hi number 6 from the spawned thread!
hi number 7 from the spawned thread!
hi number 8 from the spawned thread!
hi number 9 from the spawned thread!
这两个线程仍然会交替执行，不过主线程会由于 handle.join() 调用会等待直到新建线程执行完毕。
线程与 move 闭包
❝
move闭包，其经常与thread::spawn一起使用，因为它允许我们「在一个线程中使用另一个线程的数据」。 ❞
可以在参数列表前使用 move 关键字「强制闭包获取其使用的环境值的所有权」。
为了在新建线程中使用来自于主线程的数据，需要新建线程的闭包获取它需要的值。
代码语言：javascript复制use std::thread;
fn main() {
let v = vec![1, 2, 3];
let handle = thread::spawn(move || {
println!("Here's a vector: {:?}", v);
});
handle.join().unwrap();
}
通过告诉 Rust 将 v 的所有权移动到新建线程，我们向 Rust 保证主线程不会再使用 v。 move 关键字覆盖了 Rust 默认保守的借用，但它不允许我们违反所有权规则。
使用消息传递在线程间传送数据
确保安全并发的方式是 消息传递message passing，这里线程或 actor 通过发送包含数据的消息来相互沟通。这个思想来源于 Go 编程语言文档中 的口号：「不要通过共享内存来通讯；而是通过通讯来共享内存。」
Rust 中一个实现消息传递并发的主要工具是 通道channel，Rust 标准库提供了其实现的编程概念。
编程中的通道有「两部分组成」，一个发送者transmitter和一个接收者receiver。
- 发送者位于「上游位置」，在这里可以将橡皮鸭放入河中，
- 接收者则位于下游，橡皮鸭最终会漂流至此。
代码中的一部分调用发送者的方法以及希望发送的数据，另一部分则检查接收端收到的消息。当发送者或接收者任一被丢弃时可以认为通道被 关闭（closed）了。
use std::sync::mpsc;
fn main() {
let (tx, rx) = mpsc::channel();
}
这里使用 mpsc::channel 函数创建一个新的通道；mpsc 是 多个生产者，单个消费者multiple producer, single consumer的缩写。简而言之，Rust 标准库实现通道的方式意味着一个通道可以有多个产生值的发送sending端，但只能有一个消费这些值的接收receiving端。
❝
mpsc::channel函数返回一个元组：第一个元素是发送端，而第二个元素是接收端。 ❞
由于历史原因，tx 和 rx 通常作为 发送者（transmitter）和 接收者（receiver）的缩写，所以这就是我们将用来绑定这两端变量的名字。这里使用了一个 let 语句和模式来解构了此元组。
将发送端移动到一个新建线程中并发送一个字符串，这样新建线程就可以和主线程通讯了
代码语言：javascript复制use std::thread;
use std::sync::mpsc;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let val = String::from("hi");
tx.send(val).unwrap();
});
}
使用 thread::spawn 来创建一个新线程并使用 move 将 tx 移动到闭包中这样新建线程就拥有 tx 了。新建线程需要拥有通道的发送端以便能向通道发送消息。
通道的发送端有一个 send 方法用来获取需要放入通道的值。send 方法返回一个 Result<T, E> 类型，所以如果接收端已经被丢弃了，将没有发送值的目标，所以发送操作会返回错误。
随后，在主线程中从通道的接收端获取值。
代码语言：javascript复制use std::thread;
use std::sync::mpsc;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let val = String::from("hi");
tx.send(val).unwrap();
});
let received = rx.recv().unwrap();
println!("Got: {}", received);
}
通道的接收端有两个有用的方法：recv 和 try_recv。这里，我们使用了 recv，它是 receive 的缩写。这个方法会「阻塞主线程执行直到从通道中接收一个值」。一旦发送了一个值，recv 会在一个 Result<T, E> 中返回它。当通道发送端关闭，recv 会返回一个错误表明不会再有新的值到来了。
try_recv 「不会阻塞」，相反它「立刻返回」一个 Result<T, E>：Ok 值包含可用的信息，而 Err 值代表此时没有任何消息。如果线程在等待消息过程中还有其他工作时使用 try_recv 很有用：可以编写一个循环来频繁调用 try_recv，在有可用消息时进行处理，其余时候则处理一会其他工作直到再次检查。
通道与所有权转移
所有权规则在消息传递中扮演了重要角色，其有助于我们编写安全的并发代码。防止并发编程中的错误是在 Rust 程序中考虑所有权的一大优势。
存在如下代码：
代码语言：javascript复制use std::thread;
use std::sync::mpsc;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
let val = String::from("hi");
tx.send(val).unwrap();
println!("val is {}", val);
});
let received = rx.recv().unwrap();
println!("Got: {}", received);
}
尝试在通过 tx.send 发送 val 到通道中之后将其打印出来。
❝一旦将值发送到另一个线程后，那个线程可能会在我们再次使用它之前就将其修改或者丢弃。其他线程对值可能的修改会由于不一致或不存在的数据而导致错误或意外的结果。 ❞
Rust 会给出一个错误：
代码语言：javascript复制error[E0382]: use of moved value: `val`
--> src/main.rs:10:31
|
9 | tx.send(val).unwrap();
| --- value moved here
10 | println!("val is {}", val);
| ^^^ value used here after move
|
发送多个值并观察接收者的等待
构建一个示例：新建线程现在会发送多个消息并在每个消息之间暂停一秒钟。
代码语言：javascript复制use std::thread;
use std::sync::mpsc;
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);
}
}
在新建线程中有一个字符串 vector 希望发送到主线程。我们遍历他们，单独的发送每一个字符串并通过一个 Duration 值调用 thread::sleep 函数来暂停一秒。
在主线程中，不再显式调用 recv 函数：而是「将 rx 当作一个迭代器」。对于每一个接收到的值，我们将其打印出来。当通道被关闭时，迭代器也将结束。
将看到如下输出，「每一行都会暂停一秒」：
代码语言：javascript复制Got: hi
Got: from
Got: the
Got: thread
通过克隆发送者来创建多个生产者
mpsc 是 multiple producer, single consumer 的缩写。
代码语言：javascript复制❝可以运用
mpsc来创建「向同一接收者发送值的多个线程」。这可以通过克隆通道的发送端来做到。 ❞
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"),
];
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 方法。这会给我们一个可以传递给第一个新建线程的发送端句柄。我们会将原始的通道发送端传递给第二个新建线程。这样就「会有两个线程，每个线程将向通道的接收端发送不同的消息」。
共享状态并发
❝某种程度上，任何编程语言中的通道都类似于「单所有权」，因为「一旦将一个值传送到通道中，将无法再使用这个值」。共享内存类似于多所有权：多个线程可以同时访问相同的内存位置。 ❞
互斥器一次只允许一个线程访问数据
互斥器mutex是 mutual exclusion 的缩写，也就是说，「任意时刻，其只允许一个线程访问某些数据」。为了访问互斥器中的数据，线程首先需要通过获取互斥器的 锁lock来表明其希望访问数据。「锁是一个作为互斥器一部分的数据结构，它记录谁有数据的排他访问权」。因此，我们描述互斥器为通过锁系统 保护（guarding）其数据。
❝互斥锁有如下的规则：
- 在使用数据之前尝试获取锁。
- 处理完被互斥器所保护的数据之后，必须解锁数据，这样其他线程才能够获取锁。
❞
Mutex<T>的 API
代码语言：javascript复制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 中的值之前获取锁：Mutex<i32> 并不是一个 i32，所以「必须」获取锁才能使用这个 i32 值。
Mutex<T> 是一个「智能指针」。更准确的说，lock 调用「返回」一个叫做 MutexGuard 的智能指针。这个智能指针「实现了 Deref 来指向其内部数据」；其也提供了一个 Drop 实现当 MutexGuard 「离开作用域时自动释放锁」。为此，我们不会冒忘记释放锁并阻塞互斥器为其它线程所用的风险，因为「锁的释放是自动发生的」。
丢弃了锁之后，可以打印出互斥器的值，并发现能够将其内部的 i32 改为 6。
在线程间共享 Mutex<T>
尝试使用 Mutex<T> 在多个线程间共享值。我们将启动十个线程，并在各个线程中对同一个计数器值加一，这样计数器将从 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());
}
创建了一个 counter 变量来存放内含 i32 的 Mutex<T>。接下来遍历 range 创建了 10 个线程。使用了 thread::spawn 并对所有线程「使用了相同的闭包」：他们每一个都将调用 lock 方法来获取 Mutex<T> 上的锁，接着将互斥器中的值加一。当一个线程结束执行，num 会离开闭包作用域并释放锁，这样另一个线程就可以获取它了。
在主线程中，收集了所有的 join 句柄，调用它们的 join 方法来确保所有线程都会结束。这时，主线程会获取锁并打印出程序的结果。
编译上述代码，会发生如下错误：
代码语言：javascript复制error[E0382]: use of moved value: `counter`
--> src/main.rs:9:36
|
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
错误信息表明 counter 值在上一次循环中被移动了。所以 Rust 告诉我们不能将 counter 锁的所有权移动到多个线程中。
多线程和多所有权
通过使用智能指针 Rc<T> 来创建引用计数的值，以便拥有多所有者。将上述代码中的 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());
}
不幸的是，Rc<T> 并不能安全的在线程间共享。当 Rc<T> 管理引用计数时，它「必须」在每一个 clone 调用时增加计数，并在每一个克隆被丢弃时减少计数。Rc<T> 并「没有使用任何并发原语」，来确保改变计数的操作不会被其他线程打断。
原子引用计数 Arc<T>
所幸 Arc<T> 正是 这么一个类似 Rc<T> 并可以安全的用于并发环境的类型。字母 “a” 代表 原子性atomic，所以这是一个原子引用计数atomically reference counted类型。
Arc<T> 和 Rc<T> 有着相同的 API，所以修改程序中的 use 行和 new 调用。
use std::sync::{Mutex, Arc};
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());
}
使用 Sync 和 Send trait 的可扩展并发
通过 Send 允许在线程间转移所有权
Send 标记 trait 表明「类型的所有权可以在线程间传递」。几乎所有的 Rust 类型都是 Send 的，不过有一些例外，包括 Rc<T>：这是不能 Send 的，因为如果克隆了 Rc<T> 的值并尝试将克隆的所有权转移到另一个线程，这两个线程都可能同时更新引用计数。为此，Rc<T> 被实现为「用于单线程场景」，这时不需要为拥有线程安全的引用计数而付出性能代价。
Sync 允许多线程访问
Sync 标记 trait 表明一个实现了 Sync 的类型可以安全的在多个线程中拥有其值的引用。换一种方式来说，「对于任意类型 T，如果 &T（T 的引用）是 Send 的话 T 就是 Sync 的」，这意味着其引用就可以安全的发送到另一个线程。
后记
「分享是一种态度」。
参考资料：《Rust权威指南》
