16.2 使用消息传递来跨线程传递数据

16.2.1. 消息传递

有一种很流行而且能保证安全并发的技术叫做消息传递。在这种机制里，线程（或 Actor）通过彼此间发送消息（数据）来进行通讯。

Go 语言有一句名言是这么说的： Do not communicate by sharing memory; instead, share memory by communicating.（不要用共享内存来通信，要用通信来共享内存）

Go 语言的并发模型体现了这种思想。Rust 也提供了基于消息传递的一种并发方式，具体就是使用标准库中的 Channel。Go 语言里也有 Channel，思路差不多。

16.2.2. 理解 `Channel`

可以将编程中的 Channel 想象为定向水道，例如小溪或河流。如果你把橡皮鸭之类的东西放入河中，它会顺流而下，到达水道的尽头。

通道有两部分：发送端和接收端。发送端是将橡皮鸭放入河中的上游位置，接收端是橡皮鸭最终到达下游的位置。代码的一部分使用要发送的数据调用发送端上的方法，另一部分检查接收端是否有到达的消息。如果发送端或接收端其一消失，则称通道已关闭。

具体的步骤：

调用发送端的方法，发送数据
接收端会检查和接收到达的数据
如果发送端、接收端中的任意一端被丢弃了，那么 Channel 就关闭了。

16.2.3. 创建 `channel`

使用 mpsc::channel 函数来创建 Channel。mpsc 表示 multiple producer, single consumer（多个生产者、一个消费者），表示可以有多个发送端，但是只能有一个接收端。

调用这个函数返回一个元组，有两个元素，分别是发送端和接收端。

看个例子：

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

首先使用 mpsc::channel 函数来创建 Channel，返回的元组使用模式匹配进行解构，分别用 tx 和 rx 表示发送端和接收端。
接下来创建了一个线程，使用 move 关键字表示发送端 tx 的所有权被移至分线程内，因为线程必须拥有通道发送端的所有权才能往通道里发消息。
使用 send 方法来发送消息，返回类型是 Result 类型，如果接收端被丢弃了那么返回值就是 Err，反之就是 Ok。在这里面就简单地使用 unwrap 进行错误处理即可，这样如果接收端被丢弃就会恐慌。
接收端有两个方法来获取消息，这里使用了 recv 方法（receive 的简写）。它会一直阻塞这个线程，直到有消息被传入为止。
消息被包裹在 Result 类型中，有消息就返回 Ok，反之就是 Err，一样使用 unwrap 简单地处理错误即可。

输出：

Got: hi

发送端的 `send` 方法

send 方法的参数是想要发送的数据，返回 Result 类型。如果有问题（例如接收端已经被丢弃）就会返回 Err。

接收端的方法

recv 方法：阻止当前线程执行，直到 Channel 中有值传来，一旦收到值，就返回 Result 类型，如果发送端关闭了，就会收到 Err。
try_recv 方法：不会阻塞当前线程执行，立即返回 Result 类型，有数据到达就是 Ok 变体包裹着传过来的数据；否则就返回错误。
通常是使用循环调用来检查 try_recv 的结果。一旦有消息来了就开始处理，如果没来，那么这时候也可以执行其他指令。

16.2.4. `channel` 和所有权转移

所有权在消息传递中非常重要，它能帮你编写安全、并发的代码。

看个例子：

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

在刚才的代码上加了 println!("val is {val}"); 这句话。把值传入 send 函数后想继续在线程里使用值。

输出：

$ cargo run
   Compiling message-passing v0.1.0 (/tmp/projects/message-passing)
error[E0382]: borrow of moved value: `val`
  --> src/main.rs:10:27
   |
 8 |         let val = String::from("hi");
   |             --- move occurs because `val` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(val).unwrap();
   |                 --- value moved here
10 |         println!("val is {val}");
   |                           ^^^ value borrowed here after move

For more information about this error, try `rustc --explain E0382`.
error: could not compile `message-passing` (bin "message-passing") due to 1 previous error

错误在于借用了已经移动的值 val。它的所有权已经在传入 send 时移交出去了，所以就会报错。

下一个例子通过发送多个值来观察接收者等待的过程：

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

分线程以循环的方式发送 Vector 里的各个元素，每次发送完之后会暂停 1 秒
主线程把接收端当作一个迭代器来使用（因为实现了 Iterator trait），这样就不需要显式调用 recv 函数了。每收到一个值就将它打印出来。当发送端执行完毕被丢弃时，Channel 就关闭了，循环就不会继续。程序退出。

输出：

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

16.2.5. 通过克隆创建多个发送者

继续在上一个代码的基础上稍作修改：

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

fn main() {
    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}");
    }
}

这里多了一个分线程，现在有 2 个分线程都想要给主线程发消息，所以就需要两个发送端。针对这种情况，只需要对代表发送端的变量 tx 使用 clone 方法即可，也就是原文的 let tx1 = tx.clone(); 这一句。

输出（接收顺序不确定；以下为一次代表性的本地运行结果）：

Got: hi
Got: more
Got: from
Got: messages
Got: the
Got: for
Got: thread
Got: you

接收端收到的数据会交错来自两个发送端。

Keyboard shortcuts

Rust指南