Rust 学习笔记：通过异步实现并发

Rust 学习笔记：通过异步实现并发

在本文中，我们将重点讨论线程和 future 之间的区别。

在许多情况下，使用异步处理并发性的 API 与使用线程的 API 非常相似，但它们通常具有不同的行为，并且它们几乎总是具有不同的性能特征。

用 spawn_task 创建一个新任务

使用 thread::spawn 可以创建一个新线程，我们编写的第一个程序是在两个单独的线程上进行计数。

让我们使用 async 做同样的事情。trpl crate 提供了一个看起来与 thread::spawn API 非常相似的 spawn_task 函数，以及一个 sleep 函数，它是 thread::sleep API 的异步版本。我们可以一起使用它们来实现计数示例：

use std::time::Duration;fn main() {trpl::run(async {trpl::spawn_task(async {for i in 1..10 {println!("hi number {i} from the first task!");trpl::sleep(Duration::from_millis(500)).await;}});for i in 1..5 {println!("hi number {i} from the second task!");trpl::sleep(Duration::from_millis(500)).await;}});
}

我们使用 trpl::run 设置 main 函数，以便我们的顶级函数可以是异步的。

然后我们在该块中编写两个循环，每个循环都包含一个 trpl::sleep 调用，该调用在发送下一条消息之前等待半秒。我们将一个循环放在 trpl::spawn_task 中，另一个放在顶层的 for 循环中。我们还在 sleep 调用之后添加了一个 await。

这段代码的行为类似于基于线程的实现——包括当你运行它时，你可能会看到消息以不同的顺序出现在你自己的终端上。

这个版本在主异步块体中的 for 循环完成后立即停止，因为在主函数结束时，由 spawn_task 生成的任务被关闭。如果希望它一直运行到任务完成，则需要使用连接句柄来等待第一个任务完成。对于线程，我们使用 join 方法来“阻塞”，直到线程完成运行。我们可以使用 await 来做同样的事情，因为任务句柄本身就是一个 future。它的输出类型是 Result，所以我们也在等待它之后展开它。

use std::time::Duration;fn main() {trpl::run(async {let handle = trpl::spawn_task(async {for i in 1..10 {println!("hi number {i} from the first task!");trpl::sleep(Duration::from_millis(500)).await;}});for i in 1..5 {println!("hi number {i} from the second task!");trpl::sleep(Duration::from_millis(500)).await;}handle.await.unwrap();});
}

这个更新的版本运行直到两个循环结束。

到目前为止，看起来 async 和线程给出了相同的基本结果，只是语法不同：使用 await 而不是在连接句柄上调用 join，并等待 sleep 调用。

更大的区别在于，我们不需要生成另一个操作系统线程来执行此操作。实际上，我们甚至不需要在这里生成任务。由于 async 块编译为匿名的 future，我们可以将每个循环放在 async 块中，并让运行时使用 trpl::join 函数将它们运行到完成。

我们之前展示了如何在调用 std::thread::spawn 时对返回的 JoinHandle 类型使用 join 方法。trpl::join 函数与此类似，但用于 future。当你给它两个 future 时，它会产生一个新的 future，它的输出是一个元组，其中包含你传入的每个 future 完成后的输出。

我们使用 trpl::join 来等待 fut1 和 fut2 完成。我们不等待 fut1 和 fut2，而是等待 trpl::join 生成的新 future。

use std::time::Duration;fn main() {trpl::run(async {let fut1 = async {for i in 1..10 {println!("hi number {i} from the first task!");trpl::sleep(Duration::from_millis(500)).await;}};let fut2 = async {for i in 1..5 {println!("hi number {i} from the second task!");trpl::sleep(Duration::from_millis(500)).await;}};trpl::join(fut1, fut2).await;});
}

编译运行，我们看到两个 future 都运行到完成：

现在，每次运行的结果的顺序都完全相同，这与我们在线程中看到的非常不同。

这是因为 trpl::join 函数是公平的，这意味着它同样频繁地检查每个 future，在它们之间交替，如果另一个准备好了，它永远不会让一个抢先。对于线程，操作系统决定检查哪个线程以及让它运行多长时间。对于异步 Rust，运行时决定检查哪个任务。

在实践中，细节变得复杂，因为异步运行时可能会在后台使用操作系统线程作为管理并发性的一部分，因此保证公平性对运行时来说可能需要更多的工作。

运行时不必保证任何给定操作的公平性，它们通常提供不同的 API，让你选择是否需要公平性。

使用消息传递计算两个任务

我们使用消息传递的异步版本在 future 之间共享数据。

我们将采用与使用消息传递在线程之间传输数据略有不同的方法来说明基于线程的并发和基于 future 的并发之间的一些关键区别。

在 trpl::run 的 async 块中创建通道：

fn main() {trpl::run(async {let (tx, mut rx) = trpl::channel();let val = String::from("hi");tx.send(val).unwrap();let received = rx.recv().await.unwrap();println!("Got: {received}");});
}

这里，我们使用 trpl::channel，这是 std::mpsc::channel（多生产者、单消费者通道）的异步版本。异步版本的 API 与基于线程的版本只有一点不同：它使用一个可变的接收端 rx，它的 recv 方法产生一个我们需要等待的 future，而不是直接产生值。现在我们可以将消息从发送者发送到接收者。注意，我们不需要生成一个单独的线程或任务，我们只需要等待 rx.recv 调用。

在 std::mpsc::channel 中的 Receiver::recv 方法阻塞线程，直到它接收到消息。trpl::Receiver::recv 方法是异步的，它不阻塞，而是将控制权交还给运行时，直到接收到消息或通道的发送端关闭为止。相比之下，我们不等待 send 调用，因为它不会阻塞。

注意：由于所有这些异步代码都在 trpl::run 调用中的异步块中运行，因此其中的所有代码都可以避免阻塞。但是，它外面的代码将在运行函数返回时阻塞。这就是 trpl::run 函数的全部意义：它允许你选择在哪里阻塞某些异步代码集，以及在哪里在同步代码和异步代码之间转换。在大多数异步运行时，run 实际上被命名为 block_on 正是出于这个原因。

关于这个例子，请注意两点。首先，消息会马上到达。第二，虽然我们在这里使用了 future，但是还没有并发。程序的一切都是按顺序进行的，就像不涉及 future 一样。

让我们通过发送一系列消息并在它们之间休眠来解决第一部分：

use std::time::Duration;fn main() {trpl::run(async {let (tx, mut rx) = trpl::channel();let vals = vec![String::from("hi"),String::from("from"),String::from("the"),String::from("future"),];for val in vals {tx.send(val).unwrap();trpl::sleep(Duration::from_millis(500)).await;}while let Some(value) = rx.recv().await {println!("received '{value}'");}});
}

Rust 还没有一种方法可以在一系列异步项上编写 for 循环，因此我们需要使用 while let 条件循环，只要循环指定的模式继续匹配该值，循环就会继续执行。

rx.recv() 产生一个我们等待的 future。运行时将暂停 future，直到它准备好。一旦消息到达，future 将解析为 Some(message)。当通道关闭时，无论是否有消息到达，future 都将解析为 None，表示没有更多的值，因此我们应该停止轮询——也就是说，停止 await。

while let 循环将所有这些组合在一起。如果调用 rx.recv().await 的结果是Some(message)，则可以访问该消息，并可以在循环体中使用它。如果结果为 None，则循环结束。每次循环完成时，它都会再次到达等待点，因此运行时将再次暂停它，直到另一条消息到达。

代码现在成功地发送和接收了所有消息：

不幸的是，仍然存在一些问题。首先，消息不会以半秒的间隔到达，它们在我们启动程序后 2 秒同时到达。其次，这个程序永远不会退出！相反，它会永远等待新的消息。

因为程序中只有一个异步块，因此其中的所有内容都是线性运行的，仍然没有并发性。所有的 tx.send 调用都会发生，并与所有的 trpl::sleep 调用及其相关的等待点穿插在一起。只有这样，while let 循环才能通过 recv 调用上的任何等待点。

为了获得我们想要的行为，即在每个消息之间发生睡眠延迟，我们需要将 tx 和 rx 操作放在各自的异步块中，然后运行时可以使用 trpl::join 分别执行它们中的每一个。同样，我们等待调用 trpl::join 的结果，而不是单个的 future。

use std::time::Duration;fn main() {trpl::run(async {let (tx, mut rx) = trpl::channel();let tx_fut = async {let vals = vec![String::from("hi"),String::from("from"),String::from("the"),String::from("future"),];for val in vals {tx.send(val).unwrap();trpl::sleep(Duration::from_millis(500)).await;}};let rx_fut = async {while let Some(value) = rx.recv().await {println!("received '{value}'");}};trpl::join(tx_fut, rx_fut).await;});
}

消息以 500 ms 的间隔打印，而不是在 2 s 后匆忙打印。

然而，由于 while let 循环与 trpl::join 的交互方式，程序仍然不会退出：

• 只有当传递给它的两个 future 都完成后，从 trpl::join 返回的 future 才会完成。
• 在发送 vals 中的最后一条消息后，一旦结束 sleep，tx future 就完成了。
• 直到 while let 循环结束，rx future 才会完成。
• while let 循环直到等待 rx.recv 产生 None 才会结束。
• 等待 rx.recv 只会在通道的另一端关闭时返回 None。
• 只有当我们调用 rx.close 或当发送端 tx 被丢弃时，通道才会关闭。
• 我们不会在任何地方调用 rx.close，并且在传递给 trpl::run 的最外层异步块结束之前，tx 不会被丢弃。
• 这个块不能结束，因为它在 trpl::join 完成时被阻塞了，这将我们带回到列表的顶部。

我们可以通过在某处调用 rx.close 来手动关闭 rx，但这没有多大意义。在处理任意数量的消息后停止将使程序关闭，但我们可能会错过消息。我们需要一些其他的方法来确保 tx 在函数结束前被删除。

现在，我们发送消息的异步块只借用 tx，因为发送消息不需要所有权，但是如果我们可以将 tx 移动到异步块中，那么一旦该块结束，它就会被丢弃。move 关键字对异步块的作用就像对闭包的作用一样，将数据转移到异步块中。

我们将用于发送消息的块从 async 更改为 async move。当我们运行这个版本的代码时，它会在发送和接收最后一条消息后优雅地关闭。

use std::time::Duration;fn main() {trpl::run(async {let (tx, mut rx) = trpl::channel();let tx_fut = async move {let vals = vec![String::from("hi"),String::from("from"),String::from("the"),String::from("future"),];for val in vals {tx.send(val).unwrap();trpl::sleep(Duration::from_millis(500)).await;}};let rx_fut = async {while let Some(value) = rx.recv().await {println!("received '{value}'");}};trpl::join(tx_fut, rx_fut).await;});
}

因为 tx 所有权被转移到 async 块内，在该块执行完也就是发送作业结束之后，tx 随之被销毁，触发通道关闭，接收端返回 None。

这个异步通道也是一个多生产者通道，所以如果我们想从多个 future 发送消息，我们可以在 tx 上调用 clone。

use std::time::Duration;fn main() {trpl::run(async {let (tx, mut rx) = trpl::channel();let tx1 = tx.clone();let tx1_fut = async move {let vals = vec![String::from("hi"),String::from("from"),String::from("the"),String::from("future"),];for val in vals {tx1.send(val).unwrap();trpl::sleep(Duration::from_millis(500)).await;}};let rx_fut = async {while let Some(value) = rx.recv().await {println!("received '{value}'");}};let tx_fut = async move {let vals = vec![String::from("more"),String::from("messages"),String::from("for"),String::from("you"),];for val in vals {tx.send(val).unwrap();trpl::sleep(Duration::from_millis(1500)).await;}};trpl::join3(tx1_fut, tx_fut, rx_fut).await;});
}

克隆 tx，在第一个异步块之外创建 tx1，我们将 tx1 移动到该块中。然后将原始 tx 移动到一个新的异步块中，在那里我们以稍慢的延迟发送更多消息。

用于发送消息的两个异步块都需要是 async move 块，以便在这些块完成时丢弃 tx 和 tx1。最后，我们从 trpl::join 切换到 trpl::join3 来处理额外的 future。

现在我们看到了来自两个发送 future 的所有消息，由于发送 future 在发送后使用的延迟略有不同，因此接收消息的间隔也不同。

这是一个良好的开端，但它限制了我们的 future 数量：两个对应 join，或三个对应 join3。

后面我们将学习如何处理更多的 future。