Stream：按顺序到达的 Future

回想一下，我们在本章前面的“消息传递”一节中使用过异步信道的接收端。异步的 recv 方法会随着时间推移产生一系列值。这是一种更通用的模式的实例，称为 stream（流）。许多概念都可以很自然地表示为流：队列中逐渐变得可用的项；当完整数据集太大而无法放入计算机内存时，从文件系统中逐块增量拉取的数据块；或者随时间推移通过网络到达的数据。因为流也是 future，所以我们可以将它们与任何其他类型的 future 一起使用，并以有趣的方式组合它们。例如，我们可以将事件批量处理以避免触发过多的网络调用，为一系列长时间运行的操作设置超时，或者对用户界面事件进行节流以避免做无谓的工作。

早在第 13 章，我们在“Iterator trait 和 next 方法”一节中就见过值的序列，但迭代器与异步信道接收端之间有两个区别。第一个区别是时间：迭代器是同步的，而信道接收端是异步的。第二个区别是 API。直接使用 Iterator 时，我们调用它同步的 next 方法。而对于 trpl::Receiver 这个流，我们调用的是异步的 recv 方法。除此之外，这些 API 给人的感觉非常相似，而这种相似并非巧合。流就像是迭代的一种异步形式。不过，trpl::Receiver 是专门等待接收消息的，而通用的流 API 要宽泛得多：它像 Iterator 那样提供下一项，只不过是以异步的方式。

Rust 中迭代器与流之间的这种相似性意味着，我们实际上可以从任何迭代器创建一个流。与迭代器一样，我们可以通过调用流的 next 方法并 await 其输出来处理流，如示例 17-21 所示——不过这段代码还无法编译。

Filename: src/main.rs

extern crate trpl; // required for mdbook test

fn main() {
    trpl::block_on(async {
        let values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
        let iter = values.iter().map(|n| n * 2);
        let mut stream = trpl::stream_from_iter(iter);

        while let Some(value) = stream.next().await {
            println!("The value was: {value}");
        }
    });
}

示例 17-21：从迭代器创建流并打印其值

我们从一个数字数组开始，将其转换为迭代器，然后调用 map 将所有值翻倍。接着，我们使用 trpl::stream_from_iter 函数将迭代器转换成流。然后，我们用 while let 循环遍历流中陆续到达的项。

不幸的是，当我们尝试运行这段代码时，它无法编译，而是报告说没有可用的 next 方法：

error[E0599]: no method named `next` found for struct `tokio_stream::iter::Iter` in the current scope
  --> src/main.rs:10:40
   |
10 |         while let Some(value) = stream.next().await {
   |                                        ^^^^
   |
   = help: items from traits can only be used if the trait is in scope
help: the following traits which provide `next` are implemented but not in scope; perhaps you want to import one of them
   |
1  + use crate::trpl::StreamExt;
   |
1  + use futures_util::stream::stream::StreamExt;
   |
1  + use std::iter::Iterator;
   |
1  + use std::str::pattern::Searcher;
   |
help: there is a method `try_next` with a similar name
   |
10 |         while let Some(value) = stream.try_next().await {
   |                                        ~~~~~~~~

正如这段输出所解释的，出现这个编译错误的原因是：我们需要让正确的 trait 进入作用域，才能使用 next 方法。根据我们到目前为止的讨论，你有理由认为那个 trait 是 Stream，但实际上它是 StreamExt。Ext 是 extension（扩展）的缩写，是 Rust 社区中用一个 trait 扩展另一个 trait 的常见模式。

Stream trait 定义了一个底层接口，实际上将 Iterator 和 Future 两个 trait 组合在了一起。StreamExt 在 Stream 之上提供了一组更高层的 API，包括 next 方法以及其他类似于 Iterator trait 所提供的工具方法。Stream 和 StreamExt 尚未成为 Rust 标准库的一部分，但生态系统中的大多数 crate 都使用类似的定义。

修复这个编译错误的方法是为 trpl::StreamExt 添加一条 use 语句，如示例 17-22 所示。

Filename: src/main.rs

extern crate trpl; // required for mdbook test

use trpl::StreamExt;

fn main() {
    trpl::block_on(async {
        let values = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
        // --snip--
        let iter = values.iter().map(|n| n * 2);
        let mut stream = trpl::stream_from_iter(iter);

        while let Some(value) = stream.next().await {
            println!("The value was: {value}");
        }
    });
}

示例 17-22：成功地将迭代器用作流的基础

把所有这些部分组合在一起，这段代码就能按照我们想要的方式工作了！更重要的是，现在 StreamExt 已经在作用域中，我们可以使用它的所有工具方法，就像使用迭代器一样。

键盘快捷键

Rust 程序设计语言

Stream：按顺序到达的 Future