Rust 程序设计语言

用 Async 应用并发

在本节中，我们将把 async 应用到第 16 章中用线程处理过的一些相同的 并发挑战上。因为我们已经在那里讨论了很多关键思想，所以在本节中 我们将专注于线程和 futures 之间的不同之处。

在许多情况下，使用 async 处理并发的 API 与使用线程的 API 非常相似。 在其他情况下，它们最终会有很大不同。即使线程和 async 之间的 API 看起来相似，它们的行为也往往不同——而且它们的性能特征 几乎总是不同。

使用 spawn_task 创建新任务

我们在第 16 章「使用 spawn 创建新线程」 一节中处理的第一个操作是在两个独立的线程上计数。让我们用 async 做同样的事情。trpl crate 提供了一个 spawn_task 函数，它看起来 与 thread::spawn API 非常相似，还有一个 sleep 函数，它是 thread::sleep API 的异步版本。我们可以一起使用它们来实现 计数示例，如代码清单 17-6 所示。

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(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::block_on 设置 main 函数，这样我们的 顶层函数就可以是异步的。

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

这段代码的行为类似于基于线程的实现——包括你在自己的终端中运行时 可能看到消息以不同顺序出现这一事实：

hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!

这个版本在主 async 块中的 for 循环体完成时就停止了，因为当 main 函数 结束时，由 spawn_task 启动的任务会被关闭。如果你想让它一直运行到任务 完成，你需要使用一个 join 句柄来等待第一个任务完成。使用线程时，我们 使用 join 方法来「阻塞」直到线程运行完毕。在代码清单 17-7 中，我们 可以使用 await 做同样的事情，因为任务句柄本身就是一个 future。 它的 Output 类型是 Result，所以我们在等待后还需要对它进行 unwrap。

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(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();
    });
}

这个更新后的版本会一直运行直到两个循环都完成：

hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
hi number 6 from the first task!
hi number 7 from the first task!
hi number 8 from the first task!
hi number 9 from the first task!

到目前为止，看起来 async 和线程给了我们相似的结果，只是语法不同： 使用 await 而不是在 join 句柄上调用 join，以及等待 sleep 调用。

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

在第 16 章「等待所有线程完成」 一节中，我们展示了如何对调用 std::thread::spawn 时返回的 JoinHandle 类型使用 join 方法。trpl::join 函数类似，但用于 futures。当你给它 两个 futures 时，它会产生一个单独的新 future，其输出是一个元组， 包含你传入的每个 future 在它们都完成后的输出。因此，在代码清单 17-8 中，我们使用 trpl::join 来等待 fut1 和 fut2 都完成。 我们不等待 fut1 和 fut2，而是等待 trpl::join 产生的新 future。 我们忽略输出，因为它只是一个包含两个单元值的元组。

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(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;
    });
}

当我们运行这段代码时，我们看到两个 futures 都运行到完成：

hi number 1 from the first task!
hi number 1 from the second task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
hi number 6 from the first task!
hi number 7 from the first task!
hi number 8 from the first task!
hi number 9 from the first task!

现在，你每次都会看到完全相同的顺序，这与我们在代码清单 17-7 中使用 线程和 trpl::spawn_task 时看到的情况非常不同。这是因为 trpl::join 函数是公平的（fair），意味着它检查每个 future 的频率相同， 在它们之间交替进行，并且如果一个 future 已就绪，就永远不会让另一个 跑到前面。使用线程时，由操作系统决定检查哪个线程以及让它运行多长时间。 使用 async Rust 时，由运行时决定检查哪个任务。（在实践中，细节会变得 复杂，因为异步运行时可能在底层使用操作系统线程作为其管理并发的一部分， 所以保证公平性对运行时来说可能需要更多工作——但这仍然是可能的！） 运行时不必为任何给定的操作保证公平性，它们通常提供不同的 API 来让你 选择是否需要公平性。

尝试以下等待 futures 的变体，看看它们会做什么：

• 从任一循环或两个循环周围移除 async 块。
• 在定义每个 async 块后立即等待它。
• 只将第一个循环包裹在 async 块中，并在第二个循环的函数体之后 等待生成的 future。

作为额外的挑战，看看你能否在运行代码之前就弄清楚每种情况下的 输出会是什么！

使用消息传递在两个任务之间发送数据

在 futures 之间共享数据也会很熟悉：我们将再次使用消息传递，但这次 使用类型和函数的异步版本。我们将采用与第 16 章「用消息传递在线程间 传输数据」一节略有不同的路径，以说明基于线程的并发和 基于 futures 的并发之间的一些关键差异。在代码清单 17-9 中，我们将 只从一个 async 块开始——不像启动单独线程那样启动单独的任务。

extern crate trpl; // required for mdbook test

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let val = String::from("hi");
        tx.send(val).unwrap();

        let received = rx.recv().await.unwrap();
        println!("received '{received}'");
    });
}

这里，我们使用 trpl::channel，它是第 16 章中与线程一起使用的 多生产者、单消费者通道 API 的异步版本。API 的异步版本与基于线程的版本 只有一点不同：它使用可变的而不是不可变的接收者 rx，并且它的 recv 方法产生一个我们需要等待的 future，而不是直接产生值。 现在我们可以将消息从发送者发送到接收者。请注意，我们不必启动单独的 线程甚至任务；我们只需要等待 rx.recv 调用。

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

注意这个示例的两点。首先，消息会立即到达。其次，虽然我们在这里 使用了 future，但还没有并发。代码清单中的一切都是按顺序发生的， 就像没有涉及 futures 一样。

让我们通过发送一系列消息并在它们之间休眠来解决第一部分， 如代码清单 17-10 所示。

extern crate trpl; // required for mdbook test

use std::time::Duration;

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

除了发送消息外，我们还需要接收它们。在这种情况下，因为我们知道有 多少消息进来，我们可以通过调用四次 rx.recv().await 来手动完成。 但在现实世界中，我们通常会等待某个未知数量的消息，所以我们需要 继续等待，直到确定没有更多消息为止。

在代码清单 16-10 中，我们使用了 for 循环来处理从同步通道接收的所有 项。然而，Rust 还没有办法将 for 循环用于异步产生的一系列项， 所以我们需要使用一个以前没见过的循环：while let 条件循环。 这是我们在第 6 章「用 if let 和 let...else 进行简洁控制流」 一节中看到的 if let 结构的循环版本。只要它指定的模式继续匹配值， 循环就会继续执行。

rx.recv 调用产生一个 future，我们等待它。运行时会暂停该 future 直到它就绪。一旦消息到达，future 就会解析为 Some(message)， 有多少消息到达就解析多少次。当通道关闭时，无论是否有任何消息 已经到达，future 都会解析为 None，以表明没有更多值，因此我们应该 停止轮询——即停止等待。

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

代码现在成功地发送和接收了所有消息。不幸的是，仍然存在几个问题。 一方面，消息不是以半秒间隔到达的。它们在程序启动 2 秒（2,000 毫秒）后 一次性全部到达。另一方面，这个程序也永远不会退出！相反，它会永远 等待新消息。你需要使用 ctrl-C 来关闭它。

单个 Async 块内的代码线性执行

让我们首先研究为什么消息会在完全延迟后一次性到达，而不是每条之间 都有延迟。在给定的 async 块内，代码中 await 关键字出现的顺序 也就是程序运行时它们被执行的顺序。

代码清单 17-10 中只有一个 async 块，所以其中的一切都是线性运行的。 仍然没有并发。所有的 tx.send 调用都会发生，与所有的 trpl::sleep 调用及其相关的 await 点交错进行。只有在那之后，while let 循环才会 遍历 recv 调用上的任何 await 点。

为了得到我们想要的行为，即休眠延迟发生在每条消息之间，我们需要将 tx 和 rx 操作放在它们各自的 async 块中，如代码清单 17-11 所示。 然后运行时可以像代码清单 17-8 中那样使用 trpl::join 分别执行它们。 再次强调，我们等待的是调用 trpl::join 的结果，而不是各个 futures。 如果我们按顺序等待各个 futures，我们只会回到顺序流——这正是我们 不想做的。

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(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;
    });
}

使用代码清单 17-11 中更新后的代码，消息以 500 毫秒间隔打印， 而不是在 2 秒后一下子全部涌出。

将所有权移入 Async 块

然而，程序仍然永远不会退出，这是因为 while let 循环与 trpl::join 交互的方式：

• trpl::join 返回的 future 只有在传递给它的两个 futures 都完成后才完成。
• tx_fut future 在发送完 vals 中的最后一条消息后完成休眠时完成。
• rx_fut future 在 while let 循环结束之前不会完成。
• while let 循环在等待 rx.recv 产生 None 之前不会结束。
• 等待 rx.recv 只有在通道的另一端关闭后才会返回 None。
• 通道只有在我们调用 rx.close 或发送端 tx 被丢弃时才会关闭。
• 我们没有在任何地方调用 rx.close，而 tx 在最外层传递给 trpl::block_on 的 async 块结束之前不会被丢弃。
• 该块无法结束，因为它被阻塞在 trpl::join 完成上，这让我们 回到这个列表的顶部。

目前，我们发送消息的 async 块只是借用了 tx，因为发送消息 不需要所有权，但如果我们能将 tx 移动到那个 async 块中， 它就会在该块结束时被丢弃。在第 13 章「捕获引用或移动所有权」 一节中，你学习了如何对闭包使用 move 关键字，而且正如第 16 章 「对线程使用 move 闭包」 一节中所讨论的，我们在处理线程时经常需要将数据移入闭包。同样的 基本动态也适用于 async 块，所以 move 关键字对 async 块的作用 与对闭包的作用相同。

在代码清单 17-12 中，我们将用于发送消息的块从 async 改为 async move。

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx_fut = async move {
            // --snip--
            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;
    });
}

当我们运行这个版本的代码时，它会在最后一条消息被发送和接收后 优雅地关闭。接下来，让我们看看要从多个 future 发送数据需要 改变什么。

使用 join! 宏连接多个 Futures

这个异步通道也是一个多生产者通道，所以如果我们想从多个 futures 发送消息，可以对 tx 调用 clone，如代码清单 17-13 所示。

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(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::join!(tx1_fut, tx_fut, rx_fut);
    });
}

首先，我们克隆 tx，在第一个 async 块外创建 tx1。我们将 tx1 移入该块，就像我们之前对 tx 所做的那样。然后，我们将原始的 tx 移入一个新的 async 块，在那里我们以稍慢的延迟发送更多消息。 我们碰巧把这个新的 async 块放在接收消息的 async 块之后，但它也 同样可以放在前面。关键在于 futures 被等待的顺序，而不是它们 被创建的顺序。

两个用于发送消息的 async 块都需要是 async move 块，这样 tx 和 tx1 在那些块完成时都会被丢弃。否则，我们最终会回到一开始的 同一个无限循环中。

最后，我们从 trpl::join 切换到 trpl::join! 来处理额外的 future： join! 宏可以等待任意数量的 futures，只要我们在编译时知道 futures 的数量。我们将在本章后面讨论等待未知数量 futures 的集合。

现在我们看到来自两个发送 futures 的所有消息，并且因为发送 futures 在发送后使用略有不同的延迟，消息也以这些不同的间隔被接收：

received 'hi'
received 'more'
received 'from'
received 'the'
received 'messages'
received 'future'
received 'for'
received 'you'

我们已经探讨了如何使用消息传递在 futures 之间发送数据、async 块内的 代码如何按顺序运行、如何将所有权移入 async 块，以及如何连接多个 futures。接下来，让我们讨论如何以及为什么告诉运行时它可以切换到 另一个任务。