Rust 并发编程-多线程无畏并发

并发程序是指运行多个任务的程序（或看上去是多任务），即两个及以上的任务在重叠的时间跨度内交替运行。这些任务由线程——最小的处理单元执行。在其背后，并不完全是多任务（并行）处理，而是线程之间以普通人无法感知的速度进行上下文快速切换。很多现代应用程序都依赖于这种错觉，比如服务器可以在处理请求的同时等待其他请求。当线程间共享数据时可能会出很多问题，最常见的两种是：竞态条件和死锁。

Rust 的所有权系统以及类型安全系统是一系列解决内存安全以及并发问题的强有力工具。通过所有权以及类型检查，大多数错误发生在编译期，而非运行时错误。因此，你可以在开发时修复代码，而不是在部署到生产环境后修复代码。一旦代码可以编译了，我们就可以坚信这些代码可以正确的运行于多线程环境，而不会出现其他语言中经常出现的那些难以追踪的 bug，这就是 Rust 的 无畏并发（fearless concurrency）

多线程模型

多线程编程的风险

在大部分现代操作系统中，执行中的程序代码运行于一个 进程（process）中，操作系统则负责管理多个进程。 在程序内部，也可以拥有多个同时运行的独立部分，这些独立部分的功能被称为 线程（threads）。

将程序中的计算拆分进多个线程可以改善性能，因为程序可以同时进行多个任务，不过这也会增加复杂性。因为线程是同时运行的，所以无法预先保证不同线程中的代码的执行顺序。这会导致诸如此类的问题：

1、竞争状态（Race conditions），多个线程以不一致的顺序访问数据或资源
2、死锁（Deadlocks），两个线程相互等待对方停止使用其所拥有的资源，这会阻止它们继续运行
3、可能会发生在特定情况且难以稳定重现和修复的 bug

编程语言有一些不同的方法来实现线程。很多操作系统提供了创建新线程的 API。这种由编程语言调用操作系统 API 创建线程的模模型有时被称为 1:1，一个 OS 线程对应一个语言线程。

Rust 标准库只提供了 1:1 线程模型实现。

使用 spawn 创建新线程

use std::thread;
use std::time::Duration;

fn main() {
    // 调用 thread::spawn函数并传递一个闭包，创建一个新线程
    let thread = thread::spawn(|| {
        for i in 1..10 {
            println!("this is thread {}", i);
            thread::sleep(Duration::from_millis(1));
        }
    });

    for k in 1..5 {
        println!("this is main {}", k);
        thread::sleep(Duration::from_millis(1));
    }
}
输出：
this is main 1
this is thread 1
this is main 2
this is thread 2
this is main 3
this is thread 3
this is main 4
this is thread 4
this is thread 5

我们看到主线程执行了5次循环后退出，同时，新线程虽然创建了10次循环，但也执行了5次就退出了，当主线程结束时，新线程也会结束，而不管其是否执行完毕。

如果想让新线程执行完毕再执行主线线程，可以使用JoinHandle：

use std::thread;
use std::time::Duration;

fn main() {
    // handler 是一个拥有所有权的值
    let handler = thread::spawn(||{  
        for i in 1..10 {
            println!("this is thread {}", i);
            thread::sleep(Duration::from_millis(1));
        }
    });

    for k in 1..5 {
        println!("this is main {}", k);
        thread::sleep(Duration::from_millis(1));
    }

    handler.join().unwrap(); // 阻塞主线程退出，直到新线程执行完毕
}
输出：
this is main 1
this is thread 1
this is main 2
this is thread 2
this is main 3
this is thread 3
this is main 4
this is thread 4
this is thread 5
this is thread 6
this is thread 7
this is thread 8
this is thread 9

thread::spawn 的返回值类型是 JoinHandle。JoinHandle 是一个拥有所有权的值，当对其调用 join 方法时，它会等待其线程结束。

通过调用 handle 的 join 会阻塞当前线程直到 handle 所代表的线程结束。阻塞（Blocking） 线程意味着阻止该线程执行工作或退出。

线程和 move 闭包

通过使用move闭包来实现，把主线程的变量所有权转移到闭包里

use std::thread;

fn main() {
    let v = vec![2,4,5];
    // move 把变量 v 的所有权转移到闭包里
    let thread = thread::spawn( move || {
        println!("v is {:?}", v);
    });
}
输出：
v is [2, 4, 5]

Rust 将变量 v 的所有权移动到新建线程，这样，我们在主线程就不能再使用变量 v了（比如把变量 v drop掉），Rust 就可以保证变量v在新线程是安全的。

如果没有使用move关键字，编译会报错:

$ cargo run
error[E0373]: closure may outlive the current function, but it borrows `v`, which is owned by the current function
 --> src/main.rs:6:32
  |
6 |     let handle = thread::spawn(|| {
  |                                ^^ may outlive borrowed value `v`
7 |         println!("Here's a vector: {:?}", v);
  |                                           - `v` is borrowed here
  |
note: function requires argument type to outlive `'static`

Rust 的所有权规则又一次帮助了我们！

消息传递

Rust 中一个实现消息传递并发的主要工具是 通道（channel），一个 Rust 标准库提供了其实现的编程概念。你可以将其想象为一个水流的通道，比如河流或小溪。如果你将诸如橡皮鸭或小船之类的东西放入其中，它们会顺流而下到达下游。

通道有两部分组成，一个发送者（transmitter）和一个接收者（receiver），当发送者或接收者任一被丢弃（dropped） 时可以认为通道被关闭（closed）。

通过标准库 std::sync::mpsc 来实现，mpsc 是 多个生产者，单个消费者（multiple producer, single consumer）的缩写。

注：根据 Channel 读和写的数量，Channel 可以分为：(图片引用自陈天老师《Rust编程第一课》)

在线程之间传递消息

use std::thread;
use std::sync::mpsc;

fn main() {
    // 由于历史原因，`tx` 和 `rx` 通常作为 发送者（transmitter）和接收者（receiver）的缩写，
    // 所以这就是我们将用来绑定这两端变量的名字。
    // 这里使用了一个 `let` 语句和模式来解构了此元组；
    let (tx, rx) = mpsc::channel();
    // 使用 `move` 将 `tx` 移动到闭包中这样新建线程就拥有 `tx` 了
    thread::spawn(move || {
        // `send` 方法返回一个 `Result<T, E>` 类型
        tx.send("hello").unwrap();
    });

    // recv()这个方法会阻塞主线程执行直到从信道中接收一个值
    let msg = rx.recv().unwrap();
    println!("message is {}", msg);
}

输出：
message is hello

通道的接收端有两个有用的方法：recv 和 try_recv。这里，我们使用了 recv，它是 receive 的缩写。这个方法会阻塞主线程执行直到从通道中接收一个值。一旦发送了一个值，recv 会在一个 Result<T, E> 中返回它。当通道发送端关闭，recv 会返回一个错误表明不会再有新的值到来了。

try_recv 不会阻塞，相反它立刻返回一个 Result<T, E>：Ok 值包含可用的信息，而 Err 值代表此时没有任何消息。

由于 try_recv 不会阻塞主线程执行，新线程如果没有执行完毕就无法接收到消息，会报编译错误：

use std::thread;
use std::sync::mpsc;

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

    thread::spawn(move || {
        tx.send("hello").unwrap();
    });

    // try_recv 方法不会阻塞主线程执行，无法接收到消息，就会报编译错误
    let msg = rx.try_recv().unwrap();
    println!("message is {}", msg);
}

报错：
thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: Empty', src/libcore/result.rs:1188:5

发送多个值并观察接收者的等待

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` 函数来暂停一秒
            thread::sleep(Duration::from_secs(1));
        }
    });

    // 主线程中的 `for` 循环里并没有任何暂停或等待的代码，所以可以说主线程是在等待从新建线程中接收值
    for received in rx { // 不再显式调用 rx.recv() 函数：而是将 rx 当作一个迭代器
        println!("Got: {}", received);
    }
}

如下输出，每一行都会暂停一秒：
Got: hi
Got: from
Got: the
Got: thread

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

  use std::thread; 
  use std::sync::mpsc; 
  use std::time::Duration;
  
  fn main() {
  
    let (tx, rx) = mpsc::channel();
    // 对发送者调用了 `clone` 方法克隆一个发送者
    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 {
            // 线程1的发送端句柄使用克隆的tx1，向同一接收者rx发送值
            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 {
            // 线程2的发送端句柄使用原始的tx，向同一接收者rx发送值
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    // 两个发送者 tx 和 tx1 发送的值，都由同一个接收者 rx 接收
    for received in rx {
        println!("Got: {}", received);
    }
 }
 
可能会看到这样的输出：(每次会产生不同的输出顺序，依赖于你的系统)
Got: hi
Got: more
Got: from
Got: messages
Got: for
Got: the
Got: thread
Got: you

共享状态

共享状态或数据，意味着有多个线程同时访问相同的内存位置，Rust 通过互斥器（锁），来实现共享内存并发原语。

互斥器一次只允许一个线程访问数据

互斥器（mutex）是 「mutual exclusion」 的缩写，也就是说，任意时刻，其只允许一个线程访问某些数据。为了访问互斥器中的数据，线程首先需要通过获取互斥器的锁（lock）来表明其希望访问数据。锁是一个作为互斥器一部分的数据结构，它记录谁有数据的排他访问权。

使用标准库 std::sync::Mutex 使用互斥器：

use std::sync::Mutex;
fn main() {
    let m = Mutex::new(5);
    {
        let mut num = m.lock().unwrap();
        *num = 10; // 重新赋值
        println!("num is {}",num);
    }
    println!("m is {:?}",m);
    
}
输出：
num is 10
m is Mutex { data: 10 }

使用 lock 方法获取锁，以访问互斥器中的数据。这个调用会阻塞当前线程，直到我们拥有锁为止。

Mutex 是一个智能指针。更准确的说，lock 调用 返回 一个叫做 MutexGuard 的智能指针。这个智能指针实现了Deref来指向其内部数据；其也提供了一个Drop实现当MutexGuard离开作用域时自动释放锁。

在线程间共享 Mutex

在多线程之间共享信息，存在多个所有者同时拥有所有权，可以使用 Arc 智能指针来存放Mutex，Arc 是线程安全的，Rc 是非线程安全的，所以，不能是用 Rc ，Rc 被实现为用于单线程场景。

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

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 = Rc::clone(&counter) // 这里使用 Rc 会编译报错，因为 Rc 没有实现 Send 和 Sync trait，不能安全的在线程间传递和共享
        let counter = Arc::clone(&counter); // 将所有权移入线程之前克隆了 Arc 
        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

简单总结下：一般Rc<T>/RefCell<T>结合用于单线程内部可变性， Arc<T>/Mutex<T>结合用于多线程内部可变性。

Mutex 也有造成 死锁（deadlock） 的风险。这发生于当一个操作需要锁住两个资源而两个线程各持一个锁，这会造成它们永远相互等待。

基于 Send 和 Sync 的线程安全

Rust 中与并发相关的内容都属于标准库，而不是语言本身的内容，但是有两个并发概念是内嵌于语言中的：std::marker 中的 Sync 和 Send trait。

Send 和 Sync 作用

Send和Sync是 Rust 安全并发的重中之重，但是实际上它们只是标记特征(marker trait，该类型 trait 未定义任何行为，因此非常适合用于标记), 来看看它们的作用：

• 实现Send的类型可以在线程间安全的传递其所有权
• 实现Sync的类型可以在线程间安全的共享(通过引用，当且仅当&T是Send时，T是Sync的)

由上可知，若类型 T 的引用&T是Send，则T是Sync

实现Send和Sync的类型

在 Rust 中，几乎所有类型都默认实现了Send和Sync，意味着一个复合类型(例如结构体), 只要它内部的所有成员都实现了Send或者Sync，那么它就自动实现了Send或Sync的。，只要有一个成员不是Send或Sync，该复合类型就不是Send或Sync的。

简单总结下：

1. 实现Send的类型可以在线程间安全的传递其所有权, 实现Sync的类型可以在线程间安全的共享(通过引用)
2. 绝大部分类型都实现了Send和Sync，常见的未实现的有：裸指针、Cell、RefCell、Rc 等
3. 可以为自定义类型实现Send和Sync，但是需要unsafe代码块，必须小心维护。
4. 可以为部分 Rust 中的类型实现Send、Sync，但是需要使用newtype。

注：Cell和RefCell没实现Sync(因为UnsafeCell不是Sync)，Rc两者都没实现(因为内部的引用计数器不是线程安全的)，裸指针两者都没实现（因为它本身就没有任何安全保证）

注：手动实现 Send 和 Sync 是不安全（unsafe）的，通常并不需要手动实现 Send 和 Sync trait，实现者需要使用unsafe小心维护并发安全保证。

总结

Rust 同时提供了 async/await 和 多线程两种并发模型，要熟练使用多线程并发模型首先要掌握 Rust 的线程等相关知识，比如线程创建，线程同步，线程安全等。线程间消息传递的通道 channel，线程间共享状态的智能指针 Mutex 和 Arc。类型系统和借用检查器会确保这些场景中的代码，不会出现数据竞争和无效的引用。一旦代码可以编译了，我们就可以坚信这些代码可以正确的运行于多线程环境，而不会出现其他语言中经常出现的那些难以追踪的 bug。Sync 和 Send trait 的为多线程中数据传递和共享提供安全保证。

1. 线程模型：多线程并发编程中，需要解决竞态条件，死锁以及一些难以稳定重现和修复的 bug等问题。
2. 消息传递（Message passing）并发，其中通道（channel）被用来在线程间传递消息。
3. 共享状态（Shared state）并发，结合使用 Mutex 和 Arc，可以让多个线程访问同一片数据。
4. 线程安全：Sync 和 Send trait，能够为多线程中传递或共享的数据提供安全保障。

参考

• kaisery.github.io/trpl-zh-cn/…
• course.rs/advance/con…
• nomicon.purewhite.io/send-and-sy…
• blog.rust-lang.org/2015/04/10/…
• huonw.github.io/blog/2015/0…
• www.zhihu.com/question/30…
• zhuanlan.zhihu.com/p/266562042