The Rust programming language 读书笔记——并发

发表于 2021-07-29

| 分类于 Program

| 0

| 阅读次数: 1

字数统计: 11k

阅读时长 ≈ 0:11

并发编程（concurrent programming）允许程序中的不同部分相互独立地运行，而并行编程（parallel programming）则允许程序中的不同部分同时执行。
Rust 中的所有权和类型系统能够同时帮助开发者管理内存安全及并发问题。

高级语言往往通过放弃部分控制能力来获得有益于用户的抽象，因此只支持全部解决方案的一部分是可以理解的设计策略。比如 Erlang 提供了一套优雅的消息传递并发特性，但没有提供可以在线程间共享状态的简单方法。
底层语言被期望在任意场景下都可以提供一套性能最佳的解决方案，并对硬件建立尽可能少的抽象，因此 Rust 提供了多种建模并发问题的工具。

使用线程同时运行代码

多个线程可以同时运行，因此将程序中的计算操作拆分至多个线程可以提高性能。但也增加了程序的复杂度，因为不同线程在执行过程中的具体顺序是无法确定的。可能导致下列问题：

当多个线程以不一致的顺序访问数据或资源时产生的竞争状态（race condition）
当两个线程同时尝试获取对方持有的资源时产生的死锁（deadlock），会导致这两个线程都无法继续运行
只会出现在特定情形下且难以稳定复现和修复的 bug

使用 `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));
    }
}

程序输出如下：

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!

主线程首先打印出了文本，即便新线程的打印语句出现得更早一些。这些线程可能会交替执行，执行顺序由操作系统的线程调度策略决定。
需要注意的是，只要上述程序中的主线程运行结束，创建出的新线程也会停止，不管其打印任务是否完成。
虽然我们要求新线程不停打印文本直到 i 迭代到 9，但它在主线程停止前仅迭代到了 5。

使用 `join` 句柄等待所有线程结束

thread::spawn 的返回类型是一个自持有所有权的 joinHandle，调用它的 join 方法可以阻塞当前线程直到对应的新线程运行结束。

调用 join 方法保证新线程能够在 main 函数退出前执行完毕：

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

输出结果如下：

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!
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() 放置到 main 函数的 for 循环之前，即：

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));
        }
    });
    handle.join().unwrap();
    for i in 1..5 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }
}

在上述代码中，由于主线程会等待新线程执行完毕后才开始执行自己的 for 循环，程序的输出将不再出现交替的情形：

hi number 1 from the spawned thread!
hi number 2 from the spawned 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!
hi number 1 from the main thread!
hi number 2 from the main thread!
hi number 3 from the main thread!
hi number 4 from the main thread!

在并发编程中，诸如在哪里调用 join 等微小细节也会影响到多个线程是否能够同时运行。

在线程中使用 `move` 闭包

move 闭包常被用来与 thread::spawn 函数配合使用，允许在某个线程中使用来自另一个线程的数据。

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 错误。
Rust 在推导出如何捕获 v 后决定让闭包借用 v，因为闭包中的 println! 只需要使用 v 的引用。但 Rust 不知道新线程会运行多久，因此它无法确定 v 的引用是否一直有效。

通过在闭包前添加 move 关键字，会强制闭包获得它所需值的所有权，而不仅仅是基于 Rust 的推导来获得值的借用。

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

使用消息传递在线程间转移数据

使用消息传递（message passing）机制来保证并发安全正变得越来越流行。Go 语言文档中的口号正体现了这样的思路：不要通过共享内存来通信，而是通过通信来共享内存。

Rust 的标准库中实现了一个名为通道（channel）的编程概念，可以被用来实现基于消息传递的并发机制。
通道由发送者（transmitter）和接收者（receiver）两部分组成。发送者位于通道的上游，接收者位于下游。
某一处的代码可以通过调用发送者的方法来传送数据，另一处代码则可以通过检查接收者来获取数据。
当发送者或接收者的任何一端被丢弃，则相应的通道被关闭。

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

let received = rx.recv().unwrap();
    println!("Got: {}", received);
}

// => Got: hi

上述代码使用 mpsc::channel 函数创建了一个新的通道。mpsc 是英文 multiple producer, single consum（多个生产者，单个消费者）的缩写。
mpsc::channel 会返回一个含有发送端与接收端的元组。
再使用 thread::spawn 生成一个新线程。为了令新线程拥有发送端 tx 的所有权，使用 move 关键字将 tx 移动到了闭包的环境中。
新线程必须拥有通道发送端的所有权才能通过通道来发送消息。
发送端提供了 send 方法来处理我们想要发送的值，该方法会返回 Result<T, E> 类型作为结果。当接收端已经被丢弃而无法继续传递内容时，执行发送操作会返回一个错误。

通道的接收端有两个可用于获取消息的方法。其中 recv 会阻塞主线程的执行直到有值被传入通道。一旦有值传入通道，recv 就会将其包裹在 Result<T, E> 中返回。若通道的发送端全部关闭了，recv 会返回一个错误来表明当前通道再也没有可接收的值。
try_recv 方法不会阻塞线程，它会立即返回 Result<T, E>。当通道中存在消息时，返回包含该消息的 Ok 变体；否则返回 Err 变体。可以编写一个不断调用 try_recv 方法的循环，并在有消息时对其进行处理，没有消息时执行其他指令。

发送多个值

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

// => Got: hi
// => Got: from
// => Got: the
// => Got: thread

上述代码会迭代新线程中的动态数组来逐个发送其中的字符串，并在每次发送后调用 thread::sleep 函数来稍作暂停。
在主线程中，我们会将 rx 视作迭代器，不再显式地调用 recv 函数。迭代中的代码会打印出每个接收到的值，并在通道关闭时退出循环。
代码执行时每次打印后都会出现 1 秒的时间间隔。但我们并没有在主线程的 for 循环中执行延迟指令，表明主线程确实在等待接收新线程中传递过来的值。

通过克隆发送者创建多个生产者

通过克隆通道的发送端来创建出多个能够发送值到同一个接收端的线程：

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();

let tx1 = mpsc::Sender::clone(&tx);
    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);
    }
}

// => Got: hi
// => Got: more
// => Got: from
// => Got: messages
// => Got: for
// => Got: the
// => Got: you
// => Got: thread

共享状态的并发

从某种程度上来说，任何编程语言中的通道都有些类似于单一所有权的概念。因为用户不应该在值传递给通道后再次使用它。
基于共享内存的并发通信机制更类似于多重所有权概念，多个线程可以同时访问相同的内存地址。
我们可以通过智能指针实现多重所有权，但由于需要同时管理多个所有者，会为系统增加额外的复杂性。当然，Rust 的类型系统和所有权规则有助于正确地管理这些所有权。

互斥体（mutex）

互斥体在任意时刻只允许一个线程访问数据。为了访问互斥体中的数据，线程必须首先发出信号来获取互斥体的锁（lock）。
锁是互斥体的一部分，这种数据结构被用来记录当前谁拥有数据的唯一访问权。

关于互斥体必须牢记以下两条规则：

必须在使用数据前尝试获取锁
必须在使用完互斥体守护的数据后释放锁，这样其他线程才能继续执行获取锁的操作

在单线程环境中使用互斥体：

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5);
    {
        let mut num = m.lock().unwrap();
        *num = 6;
    }

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

// => m = Mutex { data: 6 }

为了访问 Mutex<T> 实例中的数据，我们首先需要调用其 lock 方法来获取锁。此调用会阻塞当前线程直到我们取得锁为止。
当前线程对于 lock 函数的调用会在其他某个持有锁的线程发生 panic 时失败，因此上述代码选择使用 unwrap 在意外发生时触发当前线程的 panic。

一旦获取了锁，便可以将它的返回值 num 视作一个指向内部数据的可变引用。
Rust 的类型系统会确保我们在使用 m 的值之前执行加锁操作。因为 Mutex<i32> 并不是 i32 类型，我们必须获取锁才能使用其内部的 i32 值。

实际上对 lock 的调用会返回一个名为 MutexGuard 的智能指针。该智能指针通过实现 Deref 来指向存储在内部的数据，通过实现 Drop 完成自己离开作用域时的自动解锁操作。
这种释放过程会发生在内部作用域的结尾处，因此我们不会因为忘记释放锁而导致其他线程无法继续使用该互斥体。

多个线程间共享 `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());
}

上述代码会依次启动 10 个线程，并在每个线程中分别为共享的计数器的值加 1。
但代码目前无法通过编译，会报出 error[E0382]: use of moved value: counter 错误。
原因是变量 counter 被移动进了 handle 指代的线程中，这一移动行为阻止我们在另一个线程中调用 lock 来再次捕获 counter。

多线程与多重所有权

智能指针 Rc<T> 提供的引用计数能够为单个值赋予多个所有者。
现在尝试使用 Rc<T> 来包裹 Mutex<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());
}

再次尝试编译代码，报出另外一个错误：error[E0277]: `Rc<Mutex<i32>>` cannot be sent between threads safely。
原因是 Rc<T> 在跨线程使用时并不安全。Rc<T> 会在每次调用 clone 的过程中增加引用计数，在克隆出的实例被丢弃后减少引用计数。但它并没有使用任何并发原语来保证修改计数的过程中不会被另一个线程所打断。这极有可能导致计数错误并产生诡异的 bug。

原子引用计数 `Arc<T>`

Arc<T> 类型既拥有类似于 Rc<T> 的行为，又保证自己可以被安全地用于并发场景。

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

使用 Arc<T> 替换掉代码中的 Rc<T> 后，代码可以编译通过。

需要注意的是，Rust 并不能使你完全避免使用 Mutex<T> 过程中所有的逻辑错误。使用 Mutex<T> 也会有产生死锁（deadlock）的风险。当某个操作需要同时锁住两个资源，而两个线程分别持有其中一个锁并相互请求另外一个锁时，这两个线程就会陷入无穷尽的等待过程。

The Rust Programming Language

The Rust programming language 读书笔记——并发

The Rust programming language 读书笔记——并发

使用线程同时运行代码

使用 `spawn` 创建新线程

使用 `join` 句柄等待所有线程结束

在线程中使用 `move` 闭包

使用消息传递在线程间转移数据

发送多个值

通过克隆发送者创建多个生产者

共享状态的并发

互斥体（mutex）

多个线程间共享 `Mutex<T>`

多线程与多重所有权

原子引用计数 `Arc<T>`

Recommend

一文带你理解14个Kubernetes必备基础概念

【防溯源】如何优雅的隐藏你连接 Webshell 的真实 IP

How To Create Custom Vue.js Plugins

干货｜邮件钓鱼攻击学习

What Leica-M Gear Sells The Fastest?

Facebook’s Q2 2021 revenue grows 56% despite iOS privacy changes

LinkedIn Posts Record Full-Year Revenue Result, Continues To See Strong Growth i...

Pro Tips: Pinterest Provides Insights Into How To Maximize Your Pin Strategy in...

Spark – Cannot perform Merge as multiple source rows matched… | SQL with Manoj

大基金一期赚640亿，二期即将出手！

About Joyk

The Rust programming language 读书笔记——并发

The Rust programming language 读书笔记——并发

使用线程同时运行代码

使用 spawn 创建新线程

使用 join 句柄等待所有线程结束

在线程中使用 move 闭包

使用消息传递在线程间转移数据

发送多个值

通过克隆发送者创建多个生产者

共享状态的并发

互斥体（mutex）

多个线程间共享 Mutex<T>

多线程与多重所有权

原子引用计数 Arc<T>

Recommend

About Joyk

使用 `spawn` 创建新线程

使用 `join` 句柄等待所有线程结束

在线程中使用 `move` 闭包

多个线程间共享 `Mutex<T>`

原子引用计数 `Arc<T>`