ToB企服应用市场:ToB评测及商务社交产业平台

标题: Rust 无畏并发 [打印本页]

作者: 农妇山泉一亩田 时间: 2024-5-10 04:56
标题: Rust 无畏并发
本文在原文基础上有删减，原文链接无畏并发。

目录

并发编程（Concurrent programming），代表程序的不同部分相互独立地执行，而 并行编程（parallel programming）代表程序不同部分同时执行。
注意：出于简洁的考虑将很多问题归类为并发，而不是更准确的区分并发和（或）并行，谈到并发时请自行脑内替换为并发和（或）并行。
使用线程同时运行代码

无法预先保证不同线程中的代码的执行顺序会导致诸如此类的问题：

竞态条件（Race conditions），多个线程以不一致的顺序访问数据或资源
死锁（Deadlocks），两个线程相互等待对方，这会阻止两者继续运行
只会发生在特定情况且难以稳定重现和修复的 bug

编程语言有一些不同的方法来实现线程，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));
}
}

复制代码

注意：当 Rust 程序的主线程结束时，新线程也会结束，而不管其是否执行完毕。
使用 join 等待所有线程结束

可以通过将 thread::spawn 的返回值储存在变量中来修复新建线程部分没有执行或者完全没有执行的问题，返回值类型是 JoinHandle。JoinHandle 是一个拥有所有权的值，当对其调用 join 方法时，它会等待其线程结束。
使用创建的线程的 JoinHandle 并调用 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();
}

复制代码

这两个线程会交替执行，主线程会由于 handle.join() 调用会等待直到新建线程执行完毕。
将 move 闭包与线程一同使用

move 关键字经常用于传递给 thread::spawn 的闭包，因为闭包会获取从环境中取得的值的所有权，因此会将这些值的所有权从一个线程传送到另一个线程。
使用 move 关键字强制获取它使用的值的所有权：

use std::thread;
fn main() {
let v = vec![1, 2, 3];
//Rust 不知道这个新建线程会执行多久，无法知晓对 v 的引用是否一直有效
//let handle = thread::spawn(|| {
//使用 move 关键字强制获取它使用的值的所有权
let handle = thread::spawn(move || {
println!("Here's a vector: {:?}", v);
});
// 增加 move 将不能在主线程中对其调用 drop
//drop(v);
handle.join().unwrap();
}

复制代码

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

一个日益流行的确保安全并发的方式是 消息传递（message passing），这里线程或 actor 通过发送包含数据的消息来相互沟通。

这个思想来源于 Go 编程语言文档中的口号：“不要通过共享内存来通讯；而是通过通讯来共享内存。”（“Do not communicate by sharing memory; instead, share memory by communicating.”）

为了实现消息传递并发，Rust 标准库提供了一个信道（channel）实现。信道是一个通用编程概念，表示数据从一个线程发送到另一个线程。
编程中的信息渠道（信道）有两部分组成，一个发送者（transmitter）和一个接收者（receiver）。发送者位于上游位置发送信息，收者则位于下游接受信息。代码中的一部分调用发送者的方法以及希望发送的数据，另一部分则检查接收端收到的消息。当发送者或接收者任一被丢弃时可以认为信道被关闭（closed）了。
创建一个信道并将其两端赋值给 tx 和 rx，现在还不能编译：

use std::sync::mpsc;
fn main() {.
//使用 mpsc::channel 函数创建一个新的信道
let (tx, rx) = mpsc::channel();
}

复制代码

mpsc 是 多个生产者，单个消费者（multiple producer, single consumer）的缩写。简而言之，Rust 标准库实现信道的方式意味着一个信道可以有多个产生值的发送（sending）端，但只能有一个消费这些值的接收（receiving）端。

mpsc::channel 函数返回一个元组，使用一个 let 语句和模式来解构此元组：

第一个元素是发送端 -- tx 发送者（transmitter）
第二个元素是接收端 -- rx 接收者（receiver）

将 tx 移动到一个新建的线程中并发送 “hi” ：

use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();
//创建一个新线程并使用 move 将 tx 移动到闭包中
thread::spawn(move || {
let val = String::from("hi");
//send 方法返回一个 Result<T, E> 类型
//出错的时候调用 unwrap 产生 panic，真正的程序则需要合理地处理它
tx.send(val).unwrap();
});
}

复制代码

信道的发送端有一个 send 方法用来获取需要放入信道的值，send 方法返回一个 Result 类型。
在主线程中接收并打印内容 “hi”：

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
}

复制代码

信道的接收者有两个有用的方法（recv 是 receive 的缩写）：

recv：，方法会阻塞主线程执行直到从信道中接收一个值。有发送值时 recv 会在一个 Result 中返回它，信道发送端关闭时 recv 会返回一个错误表明不会再有新的值到来了。
try_recv：不会阻塞，相反它立刻返回一个 Result：Ok 值包含可用的信息，而 Err 值代表此时没有任何消息。可以编写一个循环来频繁调用 try_recv，在有可用消息时进行处理，其余时候则处理一会其他工作直到再次检查。

信道与所有权转移

尝试在新建线程中的信道中发送完 val 值之后再使用它（代码无法通过编译）：

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);
});
let received = rx.recv().unwrap();
println!("Got: {}", received);
}

复制代码

send 函数获取其参数的所有权并移动这个值归接收者所有，这可以防止在发送后再次意外地使用这个值。
发送多个值并观察接收者的等待

发送多个消息，并在每次发送后暂停一段时间：

use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
let (tx, rx) = mpsc::channel();
thread::spawn(move || {
//字符串 vector
let vals = vec![
String::from("hi"),
String::from("from"),
String::from("the"),
String::from("thread"),
];
//遍历 vector
for val in vals {
//单独的发送每一个字符串
tx.send(val).unwrap();
//调用 thread::sleep 函数来暂停一秒
thread::sleep(Duration::from_secs(1));
}
});
//将 rx 当作一个迭代器，打印每一个接收到的值
//当信道被关闭时迭代器也将结束
for received in rx {
println!("Got: {}", received);
}
}

复制代码

输出如下：

Got: hi
Got: from
Got: the
Got: thread

复制代码

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

从多个生产者发送多个消息：

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

复制代码

共享状态并发

消息传递是一个很好的处理并发的方式，但并不是唯一一个。另一种方式是让多个线程拥有相同的共享数据。
互斥器一次只允许一个线程访问数据

互斥器（mutex）是 mutual exclusion 的缩写，任意时刻其只允许一个线程访问某些数据。为了访问互斥器中的数据，线程首先需要通过获取互斥器的 锁（lock）来表明其希望访问数据，互斥器通过锁系统保护（guarding）其数据。
Mutex的 API

出于简单的考虑，在一个单线程上下文中探索 Mutex 的 API：

use std::sync::Mutex;
fn main() {
//使用关联函数 new 来创建一个 Mutex<T>
let m = Mutex::new(5);
{
//使用 lock 方法获取锁以访问互斥器中的数据
//调用会阻塞当前线程，直到拥有锁为止
//如果另一个线程拥有锁并且那个线程 panic 了，则 lock 调用会失败，选择 unwrap 并在遇到这种情况时使线程 panic。
let mut num = m.lock().unwrap();
*num = 6;
}
println!("m = {:?}", m);
}

复制代码

lock 调用返回一个叫做 MutexGuard 的智能指针，它实现了 Deref 来指向其内部数据，也提供了一个 Drop 实现当 MutexGuard 离开作用域时自动释放锁，锁的释放是自动发生的。
在线程间共享 Mutex

程序启动了 10 个线程，每个线程都通过 Mutex 来增加计数器的值：

//这段代码无法通过编译！
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 值在上一次循环中被移动了，不能将 counter 锁的所有权移动到多个线程中。
多线程和多所有权

尝试使用 Rc 来允许多个线程拥有 Mutex

//这段代码无法通过编译！
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 并不能安全的在线程间共享，Rc 并没有使用任何并发原语来确保改变计数的操作不会被其他线程打断，在计数出错时可能会导致诡异的 bug。
原子引用计数 Arc

Arc 是一个类似 Rc 并可以安全的用于并发环境的类型，字母 “a” 代表 原子性（atomic），所以这是一个 原子引用计数（atomically reference counted）类型。

原子性是另一类这里还未涉及到的并发原语：请查看标准库中 std::sync::atomic 的文档来获取更多细节。

使用 Arc 包装一个 Mutex 能够实现在多线程之间共享所有权：

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

复制代码

RefCell/Rc 与 Mutex/Arc 的相似性

使用 RefCell 可以改变 Rc 中的内容那样，可以使用 Mutex 来改变 Arc 中的内容。
使用 Rc 有造成引用循环的风险，这时两个 Rc 值相互引用造成内存泄漏，同理 Mutex 也有造成死锁（deadlock）的风险。

使用 Sync 和 Send trait 的可扩展并发

由于不需要语言提供并发相关的基础设施，并发方案不受标准库或语言所限。然而有两个并发概念是内嵌于语言中的：std::marker 中的 Sync 和 Send trait。
通过 Send 允许在线程间转移所有权

Send 标记 trait 表明实现了 Send 的类型值的所有权可以在线程间传送，任何完全由 Send 的类型组成的类型也会自动被标记为 Send，几乎所有基本类型都是 Send 的。
Sync 允许多线程访问

Sync 标记 trait 表明一个实现了 Sync 的类型可以安全的在多个线程中拥有其值的引用，基本类型是 Sync 的，完全由 Sync 的类型组成的类型也是 Sync 的。
手动实现 Send 和 Sync 是不安全的

通常并不需要手动实现 Send 和 Sync trait，手动实现这些标记 trait 涉及到编写不安全的 Rust 代码，在创建新的由不是 Send 和 Sync 的部分构成的并发类型时需要多加小心，以确保维持其安全保证。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)