Rust 程序设计语言

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

一种日益流行的确保安全并发的方式是消息传递， 即线程或 actor 通过相互发送包含数据的消息 来进行通信。这是 Go 语言文档中一句口号所表达的理念： “不要通过共享内存来通信；相反，通过通信来共享内存。”

为了实现消息发送的并发，Rust 的标准库提供了 通道（channel）的实现。通道是一个通用的编程概念， 数据通过它从一个线程发送到另一个线程。

你可以把编程中的通道想象成一个有方向的水道， 比如小溪或河流。如果你把橡皮鸭之类的东西 放进河里，它会顺流而下，最终到达水道的尽头。

通道有两半：发送端和接收端。发送端是 你把橡皮鸭放进河流的上游位置，而 接收端是橡皮鸭顺流而下到达的位置。你的 一部分代码在发送端上调用方法来发送数据， 另一部分则检查接收端是否有消息到达。如果发送端或接收端中的 任何一半被丢弃，就称该通道被关闭（closed）了。

在这里，我们将逐步构建一个程序：让一个线程生成值并 将它们发送到通道中，另一个线程接收这些值并 打印出来。我们将使用通道在线程之间发送简单的值， 以此来说明这个特性。一旦你熟悉了这项技术，就可以 将通道用于任何需要相互通信的线程，比如 聊天系统，或者许多线程分别执行计算的某一部分并 将各部分发送给一个负责汇总结果的线程的系统。

首先，在示例 16-6 中，我们将创建一个通道，但不对它做任何事情。 注意，这段代码目前还不能编译，因为 Rust 无法判断我们 想通过通道发送什么类型的值。

use std::sync::mpsc;

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

我们使用 mpsc::channel 函数创建一个新通道；mpsc 是 multiple producer, single consumer（多生产者，单消费者）的缩写。 简而言之，Rust 标准库实现通道的方式意味着一个通道可以有多个 发送端来产生值，但只有一个接收端来消费这些值。想象一下 多条小溪汇入一条大河：从任何一条小溪发送下去的所有 东西最终都会汇入同一条河。我们先从单个 生产者开始，等这个示例跑通后我们再添加多个生产者。

mpsc::channel 函数返回一个元组，其第一个元素是 发送端——即 transmitter（发送器），第二个元素是接收 端——即 receiver（接收器）。tx 和 rx 这两个缩写在 许多领域习惯上分别用来表示 transmitter 和 receiver， 因此我们用这样的变量名来表示每一端。我们使用了带有 解构元组模式的 let 语句；我们将在第 19 章讨论 let 语句中模式的使用和解构。现在只需知道，以这种方式使用 let 语句是提取 mpsc::channel 返回的元组各个部分的便捷方式。

让我们把发送端移动到一个派生线程中，让它发送一个 字符串，这样派生线程就在与主线程通信了， 如示例 16-7 所示。这就像在河流上游放一只橡皮鸭， 或者从一个线程向另一个线程发送一条聊天消息。

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

我们再次使用 thread::spawn 创建一个新线程，然后使用 move 将 tx 移动到闭包中，让派生线程拥有 tx。派生 线程需要拥有发送端，才能通过通道发送消息。

发送端有一个 send 方法，接收我们想要发送的值。 send 方法返回一个 Result<T, E> 类型，因此如果接收端已经 被丢弃、没有地方可以发送值，send 操作将 返回一个错误。在这个示例中，我们调用 unwrap，在出错时 panic。 但在真实应用中，我们应该妥善处理它：请回到 第 9 章回顾正确的错误处理策略。

在示例 16-8 中，我们将在主线程中从接收端获取值。这 就像在河流尽头从水中取回橡皮鸭， 或者接收一条聊天消息。

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

接收端有两个有用的方法：recv 和 try_recv。我们使用 recv， 它是 receive（接收）的缩写，它会阻塞主线程的执行并等待， 直到有值通过通道发送过来。一旦有值被发送，recv 会 将其包装在 Result<T, E> 中返回。当发送端关闭时，recv 会返回 一个错误，表示不会再有更多值到来。

try_recv 方法不会阻塞，而是立即返回一个 Result<T, E>： 如果有可用的消息，则返回包含消息的 Ok 值；如果这次没有 任何消息，则返回 Err 值。如果这个线程在等待消息的同时 还有其他工作要做，使用 try_recv 就很有用：我们可以写一个 循环，时不时地调用 try_recv，有消息就处理， 否则先做一会儿其他工作，直到再次检查。

为了简单起见，本例中我们使用了 recv；主线程除了 等待消息之外没有其他工作要做，所以阻塞主线程是合适的。

当我们运行示例 16-8 中的代码时，将看到主线程打印出 这个值：

Got: hi

完美！

通过通道转移所有权

所有权规则在消息发送中扮演着至关重要的角色，因为它们能帮助你 编写安全的并发代码。在整个 Rust 程序中贯彻所有权思维， 其好处就在于能防止并发编程中的错误。让我们做 一个实验，展示通道与所有权如何协同工作以防止 问题：我们将尝试在派生线程中把 val 值通过通道发送之后 再使用它。试着编译示例 16-9 中的代码，看看为什么 这样的代码是不被允许的。

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

这里，我们尝试在通过 tx.send 把 val 发送到通道之后再打印它。 允许这样做将是个坏主意：一旦值被发送到另一个 线程，那个线程就可能在我们再次尝试使用该值之前修改或丢弃它。 潜在地，另一个线程的修改可能会由于数据不一致或不存在而导致错误或 意外结果。然而，如果我们尝试编译示例 16-9 中的代码，Rust 会 给我们一个错误：

$ cargo run
   Compiling message-passing v0.1.0 (file:///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

我们的并发错误导致了一个编译时错误。send 函数 获取其参数的所有权，当值被移动后，接收端 就获得了它的所有权。这阻止我们在发送值之后不小心再次使用它； 所有权系统会检查一切是否没有问题。

发送多个值

示例 16-8 中的代码编译并运行了，但它并没有清楚地展示 两个独立的线程正在通过通道相互通信。

在示例 16-10 中，我们做了一些修改，将证明示例 16-8 中的代码正在并发运行：派生线程现在将发送多条 消息，并在每条消息之间暂停一秒。

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。我们对其进行迭代，逐个发送，并通过 以一秒的 Duration 值调用 thread::sleep 函数来在每次发送之间 暂停。

在主线程中，我们不再显式调用 recv 函数： 而是把 rx 当作一个迭代器。对于接收到的每个值，我们都 打印它。当通道关闭时，迭代将结束。

运行示例 16-10 中的代码时，你应该会看到以下输出， 每行之间间隔一秒：

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

由于主线程中的 for 循环里没有任何暂停或延迟的代码， 我们可以断定主线程正在等待接收来自派生线程的值。

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

前面我们提到 mpsc 是 multiple producer, single consumer（多生产者，单消费者）的缩写。让我们运用 mpsc，扩展示例 16-10 中的代码， 创建多个都向同一个接收端发送值的线程。我们可以 通过克隆发送端来做到这一点，如示例 16-11 所示。

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

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

这一次，在创建第一个派生线程之前，我们对 发送端调用了 clone。这将给我们一个新的发送端，可以传递给第一个 派生线程。我们把原始发送端传给第二个派生线程。 这样我们就有了两个线程，各自向同一个接收端发送不同的消息。

当你运行这段代码时，输出应该类似这样：

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

取决于你的系统，你可能会看到不同顺序的值。这正是 并发既有趣又困难的地方。如果你试验 thread::sleep，在不同的线程中给它不同的值，每次运行 将更加不确定，每次都会产生不同的输出。

现在我们已经了解了通道的工作方式，接下来看看另一种 并发方式。