document/rust-lang-zh_CN
2
0
Fork 0
mirror of https://github.com/gnu4cn/rust-lang-zh_CN.git synced 2026-08-19 04:33:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Updated 'src/async/streams.md'.

Browse Source
This commit is contained in:
Hector PENG
2026-04-12 11:14:38 +08:00
parent fa5fa2c297
commit 33574b036d
3 changed files with 52 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
projects/stream_demo/Cargo.toml Normal file
View File

@@ -0,0 +1,7 @@
[package]
name = "stream_demo"
version = "0.1.0"
edition = "2024"
[dependencies]
trpl = "0.3.0"

17
projects/stream_demo/src/main.rs Normal file
View File

@@ -0,0 +1,17 @@
use trpl::StreamExt;
use std::time::Duration;
fn main() {
let fut = 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!("值为： {value}");
trpl::sleep(Duration::from_secs(1)).await;
}
};
trpl::block_on(fut)
}

468
src/async/streams.md
View File

@@ -1,17 +1,12 @@
# 流：序列中的未来值
# 流：序列未来值
**Streams：Futures in Sequence**
回顾在这一章早先 [使用消息传递在两个任务之间发送](./concurrency_n_async.md#使用消息传递在两个任务之间发送) 小节中，我们对异步信道使用接收器的方式。异步的 `recv` 方法会随着时间推移生成项目序列。这属于一种称为 *流* 的更通用模式的实例。许多概念都自然地以流的形式呈现：项目以队列形式变得可用、当完整数据集相对计算机内存过大而从文件系统中增量提取的数据块，或是随时间推移通过网络到达的数据。由于流属于未来值，我们可以像使用任何其他类型的未来值一样使用他们，并以有趣的方式组合他们。例如，我们可以批量处理事件以避免触发过多的网络调用，可以对长时间运行的操作序列设置超时，或者对用户接口事件限流，以避免执行不必要的工作。
我们在第 13 章中 [`Iterator` 特质与 `next` 方法](../functional_features/iterators.md#iterator-特质与-next-方法) 小节研究迭代器时，曾见到过条目序列，但迭代器与异步信道接收器之间有两个区别。第一个区别是时间：迭代器是同步的，而信道接收器是异步的。第二个区别是 API。在直接使用 `Iterator` 时，我们调用他的同步的 `next` 方法。而特别是对于 `trpl::Receiver` 流，我们调用他的异步的 `recv` 方法。除此之外，这些 API 感觉非常相似，而这种相似性并非偶然。流就像迭代的异步形式。尽管 `trpl::Receiver` 专门等待接收消息，而通用的流 API 则范围更广：他以 `Iterator` 的方式提供下一项目，但以异步方式。
本章到目前为止，我们主要着重于单个的未来值。唯一例外是我们曾用到的异步通道。回顾在本章前面的 [“消息传递”](concurrency_n_async.md#使用消息传递在两个任务上计数) 小节，我们使用咱们异步通道接收器的方式。那个异步 `recv` 方法，会随着时间推移产生一个条目序列。这是种所谓 *流* 的更通用模式的一个实例。
早在第 13 章 [“`Iterator` 特质与 `next` 方法”](../functional_features/iterators.md#iterator-特质与-next-方法) 小节，介绍 `Iterator` 这个特质时，我们就见到过条目序列，但迭代器与那个异步通道接收器，有两个不同点。第一个区别是时间：迭代器是同步的，而这个通道接收器是异步的。第二个区别是 API。在直接使用 `Iterator` 时，我们会调用其同步的 `next` 方法。而在这个特定的 `trpl::Receiver` 流下，我们调用了一个异步的 `recv` 方法。除此之外，这两个 API 感觉非常相似，而这种相似并非巧合。所谓流，就像迭代的一种异步形式。不过，尽管 `trpl::Receiver` 专门等待接收消息，单通用的流 API 则有着更广泛的范围：他以 `Iterator` 同样方式，提供下一项目，不过是异步地提供。
Rust 中迭代器与流之间的相似性，意味着我们实际上可以从任何迭代器，创建出一个流。与某个迭代器一样，通过调用某个流的 `next` 方法然后等待输出，我们就可以使用这个流，如下清单 17-30 所示。
Rust 中迭代器与流之间的相似性，意味着我们实际上可以从任何迭代器创建流。与迭代器一样，我们通过调用流的 `next` 方法然后等待输出来使用流，如下清单 17-30 中所示，其尚不会编译。
<a name="listing_17-30"></a>
文件名：`src/main.rs`
```rust
@@ -20,22 +15,20 @@ Rust 中迭代器与流之间的相似性，意味着我们实际上可以从任
let mut stream = trpl::stream_from_iter(iter);
while let Some(value) = stream.next().await {
println!("The value was: {value}");
println!("值为： {value}");
}
```
**清单 17-30**：从迭代器创建流，并打印其值
*清单 17-30：从某个迭代器创建出一个流，并打印其值*
我们从一个数字的数组开始，将其转换为一个迭代器，然后调用 `map` 将所有值加倍。然后我们使用 `trpl::stream_from_iter` 函数，将这个迭代器转换为一个流。接下来，我们以那个 `while let` 循环，在该流中项目到达时遍历他们。
不幸的是，当我们尝试运行该代码时，他不会编译，而是报告没有 `next` 方法可用：
我们以一个数字数组开始，将其转换为迭代器，然后调用 `map` 将所有值翻倍。然后我们使用 `trpl::stream_from_iter` 函数转换该迭代器为流。接下来，我们以 `while let` 循环在该流中的项目到达时遍历他们。
不幸的是，当我们尝试运行代码时，他不会编译，而是报告没有可用的 `next` 方法：
```console
error[E0599]: no method named `next` found for struct `tokio_stream::iter::Iter` in the current scope
$ cargo run
Compiling stream_demo v0.1.0 (/home/hector/rust-lang-zh_CN/projects/stream_demo)
error[E0599]: no method named `next` found for struct `tokio_stream::iter::Iter<I>` in the current scope
--> src/main.rs:7:40
|
7 | while let Some(value) = stream.next().await {
@@ -46,449 +39,44 @@ help: the following traits which provide `next` are implemented but not in scope
|
1 + use futures_util::stream::stream::StreamExt;
|
1 + use std::iter::Iterator;
|
1 + use std::str::pattern::Searcher;
|
1 + use trpl::StreamExt;
|
help: there is a method `try_next` with a similar name
|
7 | while let Some(value) = stream.try_next().await {
| ~~~~~~~~
| ++++
For more information about this error, try `rustc --explain E0599`.
error: could not compile `stream_demo` (bin "stream_demo") due to 1 previous error
```
正如此输出所解释的，这条编译器报错的原因，是我们需要在作用域中的正确特质，才能使用这个 `next` 方法。根据我们到目前为止的讨论，咱们可能会合理地认为，这个特质是 `Stream`，但他实际上是 `StreamExt`。`Ext` 是 `extension` 的缩写，是 Rust 社区中用于表示以一个特质，扩展另一特质的常见模式。
正如这一输出所解释的，编译器报错的原因是，我们需要作用域中的正确特质，才能使用 `next` 方法。根据我们到目前为止的讨论，咱们可能会合理地认为该特质是 `Stream`，但他实际上是 `StreamExt`。`Ext` 是 *extension* 的缩写，属于 Rust 社区中用于以一个特质扩展另一特质的常见模式。
`Stream` 特质定义了一个底层接口，其有效地结合了 `Iterator` 和 `Future` 特质。`StreamExt` 在 `Stream` 之上提供了一组更高层次的 API，包括 `next` 方法以及类似于 `Iterator` 特质提供的其他实用方法。`Stream` 和 `StreamExt` 目前均尚不属于 Rust 标准库的一部分，但大多数生态系统代码箱都使用类似的定义。
这一编译器报错的解决方法是，添加一个 `trpl::StreamExt` 的 `use` 语句，如下清单 17-22 中所示。
在本章末尾，我们将对 `Stream` 与 `StreamExt` 两个特质进行更详细的解释，但现在咱们只需知道，`Stream` 特质定义了个可有效地结合 `Iterator` 与 `Future` 特质的底层接口。而 `StreamExt` 则在 `Stream` 之上，提供了一组更高级别的 API，包括 `next` 方法以及与由 `Iterator` 特质所提供的类似其他工具方法。`Stream` 与 `StreamExt` 还不是 Rust 标准库的一部分，但大多数生态代码箱，都使用了这同样的定义。
修复这个编译器报错的方法，就是添加一条 `trpl::StreamExt` 的 `use` 语句，如下清单 17-31 所示。
<a name="listing_17-30"></a>
文件名：`src/main.rs`
```rust
use trpl::StreamExt;
fn main() {
trpl::run( async {
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-31：成功将某个迭代器，用作一个流的基础*
将所有这些部分放在一起，这段代码就可以按照我们想要的方式运行了！此外，既然咱们在作用域中有了 `StreamExt`，我们就可以像使用迭代器一样，使用 `StreamExt` 的所有工具方法。例如，在下面的清单 17-32 中，我们就使用了 `filter` 方法，过滤掉了除 3 与 5 倍数之外的所有内容。
文件名：`src/main.rs`
```rust
use trpl::StreamExt;
fn main() {
trpl::run( 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);
let mut filtered =
stream.filter(|value| value % 3 == 0 || value % 5 == 0);
while let Some(value) = filtered.next().await {
println!("The value was: {value}");
}
});
}
```
*清单 17-32：以 `StreamExt:filter` 方法过滤某个流*
当然，这并不是很有趣，因为我们可以用一般迭代器完成同样的事情，而根本不需要任何的异步。我们来看看我们能做些什么 *流* 特有的事情。
## 合成流
**Composing Streams**
许多概念都可很自然地表示为流：某个队列中的项目成为可用；当整个数据集相对于内存来说太大时，从文件系统增量地拉取数据块；或者随着时间推移，数据经由网络到达等等。由于流是一些未来值，我们可以将他们与任何的其他类型未来值一起使用，并以有趣方式将他们结合。例如，我们可以批量处理事件，避免触发过多网络调用；对长期运行的操作序列设置超时；或为用户界面事件设置阈值，避免执行不必要的工作。
我们以创建一个小的消息流，作为咱们可能在 WebSocket 或其他实时通信协议中，所见到的数据流替身，如下清单 17-33 所示。
文件名：`src/main.rs`
```rust
use trpl::{ReceiverStream, Stream, StreamExt};
fn main() {
trpl::run(async {
let mut messages = get_messages();
while let Some(message) = messages.next().await {
println!("{message}");
}
});
}
fn get_messages() -> impl Stream<Item = String> {
let (tx, rx) = trpl::channel();
let messages = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"];
for message in messages {
tx.send(format!("Message: '{message}'")).unwrap();
}
ReceiverStream::new(rx)
}
```
*清单 17-33：将 `rx` 接收器，用作一个 `ReceiverStream`*
首先，我们创建了个返回 `impl Stream<Item = String>`，名为 `get_messages` 的函数。对于其实现，我们创建了个异步通通道，遍历英语字母表的前 10 个字母，并将他们发送到该通道上。
我们还使用了一种将 `trpl::channel` 中的 `rx` 接收器，转换为有着 `next` 方法流的新类型 `ReceiverStream`。回到 `main`，我们使用一个 `while let` 循环，打印出该流中的所有消息。
当我们运行这段代码时，我们会得到正是我们所期望的结果：
```console
Message: 'a'
Message: 'b'
Message: 'c'
Message: 'd'
Message: 'e'
Message: 'f'
Message: 'g'
Message: 'h'
Message: 'i'
Message: 'j'
```
同样，我们可以使用常规的 `Receiver` API，或甚至常规的 `Iterator` API 完成这一点，不过，我们来添加一个需要流的功能：添加一个应用到流上每个项目的超时，以及一个我们发出项目上的延迟，如下清单 17-34 所示。
文件名：`src/main.rs`
```rust
use std::{pin::pin, time::Duration};
use trpl::{ReceiverStream, Stream, StreamExt};
fn main() {
trpl::run(async {
let mut messages =
pin!(get_messages().timeout(Duration::from_millis(200)));
while let Some(result) = messages.next().await {
match result {
Ok(message) => println!("{message}"),
Err(reason) => eprintln!("Problem: {reason:?}"),
}
}
})
}
```
*清单 17-34：使用 `StreamExt::timeout` 方法，在流中项目上设置一个时间限制*
我们以使用来自 `StreamExt` 特质的 `timeout` 方法，为流添加超时开始。然后，我们更新那个 `while let` 循环的主体，因为该流现在会返回一个 `Result`。其中的 `Ok` 变种表示有消息及时到达了；`Err` 变种表示在有消息到达前超时了。我们对该结果进行 `match`，要么在成功接收消息时打印出该消息，否则打印一条超时的通知。最后，请注意在对消息应用超时后，我们会对其进行了固定，因为这个超时助手会产生一个需要被固定才能轮询的流。
不过，由于消息之间没有延迟，这个超时没有改变该程序的行为。我们来为咱们发送的信息，添加一个可变延迟，如下清单 17-35 所示。
文件名：`src/main.rs`
```rust
fn get_messages() -> impl Stream<Item = String> {
let (tx, rx) = trpl::channel();
trpl::spawn_task(async move {
let messages = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"];
for (index, message) in messages.into_iter().enumerate() {
let time_to_sleep = if index % 2 == 0 { 100 } else { 300 };
trpl::sleep(Duration::from_millis(time_to_sleep)).await;
tx.send(format!("Message: '{message}'")).unwrap();
}
});
ReceiverStream::new(rx)
}
```
*清单 17-35：在不将 `get_messages` 函数构造为异步函数下，以一个异步的延迟通过 `tx` 发送消息*
在 `get_messages` 中，我们对 `messages` 这个数组，使用了迭代器方法 `enumerate`，这样咱们就能获取到咱们正发送的每个项目索引及该项目本身。然后，我们对偶数索引的项目应用 100 毫秒的延迟，对奇数索引的项目应用 300 毫秒的延迟，模拟现实世界中我们可能见到的消息流不同延迟。由于我们的超时为 200 毫秒，这应该会影响到一半的消息。
要在 `get_messages` 函数中的消息间无阻塞地睡眠，我们需要用到异步。但是，我们不能让 `get_messages` 本身构造为一个异步函数，因为这样我们就会返回一个 `Future<Output = Stream<Item = String>>` 而不是 `Stream<Item=String>>`。调用者也将必须等待 `get_messages` 本身才能访问到该流。但请记住：给定未来值中的所有项目，都是线性发生的；并发会发生在未来值 *之间*。等待 `get_messages` 就将要求他，在返回这个接收器流前发送所有消息，包括每条消息间的睡眠延迟。因此，其中的超时将毫无用处。在该流本身中，不会有延迟；所有延迟都应在该流可用前发生。
相反，我们将 `get_messages` 作为一个返回流的常规函数，且咱们生成了个任务，处理其中异步的 `sleep` 调用。
> **注意**：以这种方式调用 `spawn_task` 会工作，是因为我们已经设置了咱们的运行时；若咱们未设置运行时，其将造成一个运行时不可恢复错误，a panic。其他（译注：异步运行时）实现，选择了不同折衷方法：他们可能会生成一个新的运行时而避免这种不可恢复错误，但最终会带来一些额外的开销；或者他们会在没有到运行时的引用时，简单地不提供生成任务的独立方法。请确保咱们清楚咱们的运行时，选择了哪种折衷方法，并据此编写代码！
现在，我们的代码有了更有趣的结果。在每对信息之间，都有个 `Problem: Elapsed(())` 报错。
```console
Message: 'a'
Problem: Elapsed(())
Message: 'b'
Message: 'c'
Problem: Elapsed(())
Message: 'd'
Message: 'e'
Problem: Elapsed(())
Message: 'f'
Message: 'g'
Problem: Elapsed(())
Message: 'h'
Message: 'i'
Problem: Elapsed(())
Message: 'j'
```
超时并未阻止信息最终到达。我们仍能收到全部原始消息，因为我们的通道 *不受限制*：内存能容纳多少消息，通道就能容纳多少消息。若超时前消息没有到达，我们的流处理程序将考虑到这点，但当他再次轮询该流时，信息就可能已经到达了。
在需要时，咱们可通过使用更通用的其他类型通道或其他类型流，获得不同行为。我们来通过将一个时间间隔流与这个消息流结合，看看其中的一种实际应用。
## 合并流
**Merging Streams**
首先，我们来创建另一个，在咱们让他直接运行时，他将每毫秒发出一个条目的流。为简单起见，我们可使用 `sleep` 函数，以一定延迟发送一条消息，并将其与咱们在 `get_messages` 中使用的自通道创建出流的同样方法结合。不同的是，这次我们将发回已经历的时间间隔计数，因此返回类型将是 `impl Stream<Item = u32>`，我们可调用函数 `get_intervals`（参见清单 17-36）。
文件名：`src/main.rs`
```rust
fn get_intervals() -> impl Stream<Item = u32> {
let (tx, rx) = trpl::channel();
trpl::spawn_task(async move {
let mut count = 0;
loop {
trpl::sleep(Duration::from_millis(1)).await;
count += 1;
tx.send(count).unwrap();
}
});
ReceiverStream::new(rx)
}
```
*清单 17-36：以一个每毫秒都将被发射的计数器，创建出一个流*
我们以在该任务中定义一个 `count` 开始。(我们也可在该任务外定义它，但限制任何给定变量的作用域会更加清晰。）然后，我们创建了个无限循环。该循环的每次迭代，都会异步地休眠一毫秒，递增那个计数，然后将其发送到通道上。由于这都封装在由 `spawn_task` 创建的任务中，因此包括无限循环在内的所有内容，都将与运行时一起被清理。
在异步的 Rust 中，这种只会在整个运行时被破坏时，才结束的无限循环相当常见：许多程序都需要无限期地运行下去。在异步下，只要在该循环的每次迭代中，至少有一个等待点，就不会阻塞其他任何事情。
现在，回到咱们主函数的异步代码块，我们可以尝试合并 `messages` 与 `intervals` 两个流，如下清单 17-37 所示。
文件名：`src/main.rs`
```rust
let messages = get_messages().timeout(Duration::from_millis(200));
let intervals = get_intervals();
let merged = messages.merge(intervals);
```
*清单 17-37：尝试合并 `messages` 与 `intervals` 两个流*
我们以调用 `get_intervals` 开始。然后，我们以 `merge` 方法，合并 `messages` 与 `intervals` 两个流，该方法会将多个流合并为一个，在任一来源流中项目可用时，就会立即产生项目，而不会强加任何特定顺序。最后，我们会对合并后的流，而不再对 `messages` 循环。
此刻，`messages` 与 `intervals` 二者都不需要固定或可变，因为二者都将被合并为单一的 `merged` 流。然而，这个到 `merge` 的调用不会编译！(`while let` 循环中的 `next` 调用也不会编译，不过我们会再讨论这个问题。）这是因为两个流的类型不同。`messages` 流的类型是 `Timeout<impl Stream<Item = String>>`，其中 `Timeout` 是实现了 `Stream` 的某个 `timeout` 调用的类型。`intervals` 流的类型为 `impl Stream<Item = u32>`。要合并这两个流，我们需要转换其中一个，以匹配另一个。我们将重新设计间隔流，因为消息流已经是我们想要的基本格式，并且必须要处理超时错误（请参阅清单 17-38）。
文件名：`src/main.rs`
```rust
let messages = get_messages().timeout(Duration::from_millis(200));
let intervals = get_intervals()
.map(|count| format!("Interval: {count}"))
.timeout(Duration::from_secs(10));
let merged = messages.merge(intervals);
let mut stream = pin!(merged);
```
*清单 17-38：将 `intervals` 流的类型与 `messages` 流的类型对齐*
首先，我们可使用 `map` 这个辅助方法，将 `intervals` 转换为字符串。其次，我们需要与 `messages` 中的 `Timeout` 匹配。不过，由于我们实际上并不 *想要* `intervals` 的某个超时，因此我们可以创建一个比我们所使用的其他持续时间，更长的一个超时。在这里，我们使用 `Duration::from_secs(10)` 创建了一个 10 秒的超时。最后，我们需要将 `stream` 可变，这样 `while let` 循环的 `next` 调用，就可以遍历这个流，并将其固定，确保这样做是安全的。这样就 *差不多* 达到了我们需要的效果。其中一切都通过了类型检查。在咱们运行这个程序时，会有两个问题。首先，他永远不会停止！咱们需要用 `ctrl-c` 停止他。其次，英文字母的消息，会埋没在全部间隔计数器消息中间。
```console
--snip--
Interval: 38
Interval: 39
Interval: 40
Interval: 41
Interval: 42
Interval: 43
Interval: 44
Interval: 45
Interval: 46
Interval: 47
Message: 'a'
Interval: 48
Interval: 49
Interval: 50
--snip--
```
下面清单 17-39 给出了解决这最后两个问题的一种方式。
文件名：`src/main.rs`
```rust
let messages = get_messages().timeout(Duration::from_millis(200));
let intervals = get_intervals()
.map(|count| format!("Interval: {count}"))
.throttle(Duration::from_millis(100))
.timeout(Duration::from_secs(10));
let merged = messages.merge(intervals).take(20);
let mut stream = pin!(merged);
```
*清单 17-39：使用 `throttle` 与 `take` 管理合并的两个流*
首先，我们在 `intervals` 上使用了 `throttle` 方法，这样他就不会淹没 `messages` 流。所谓 *节流*，是种限制函数调用速率的方式 -- 或者说，在本例中，就是限制对该流轮询的频率。每 100 毫秒轮询一次就可以了，因为我们的消息大概就是以这个频率到达的。
要限制我们从某个流中接受项目的数量，我们对 `merged` 流应用了 `take` 方法，因为我们要限制的是最终输出，而不仅仅是其中一个流或另一个流。
现在，当我们运行这个程序时，他会在从该流中拉取 20 个条目后停止，且间隔时间不会淹没消息。我们也不会得到 `Interval: 100` 或 `Interval: 200` 等消息，而是得到 `Interval: 1`、`Interval: 2` 等消息 -- 尽管我们的源流，可以每毫秒产生一个事件。这是因为其中的 `throttle` 调用，会产生一个封装了原始流的新流，这样原始流就只能以节流速率，而不是其自己 “原生” 速率被轮询。我们没有了一堆的我们选择忽略的那些未处理间隔消息。相反，我们从一开始，就不会产生出这些间隔消息！这是 Rust 未来值固有的 “懒惰” 再次发挥了作用，允许我们选择一些性能特性。
```console
Interval: 1
Message: 'a'
Interval: 2
Interval: 3
Problem: Elapsed(())
Interval: 4
Message: 'b'
Interval: 5
Message: 'c'
Interval: 6
Interval: 7
Problem: Elapsed(())
Interval: 8
Message: 'd'
Interval: 9
Message: 'e'
Interval: 10
Interval: 11
Problem: Elapsed(())
Interval: 12
```
我们还需要处理最后一件事：出错！对于这两个基于通道的流，当通道的一侧关闭时，`send` 调用可能会失败 -- 这只是运行时执行组成流的未来值方式的问题。到目前为止，通过调用 `unwrap`，我们忽略了这种可能性，但在行为完备的应用中，我们应该显式地处理这个出错，至少要结束循环，从而咱们不再尝试发送任何消息。下面清单 17-40 展示了一种简单的出错策略：打印出问题，然后从循环中 `break`。
```rust
fn get_messages() -> impl Stream<Item = String> {
let (tx, rx) = trpl::channel();
trpl::spawn_task(async move {
let messages = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"];
for (index, message) in messages.into_iter().enumerate() {
let time_to_sleep = if index % 2 == 0 { 100 } else { 300 };
trpl::sleep(Duration::from_millis(time_to_sleep)).await;
if let Err(send_error) = tx.send(format!("Message: '{message}'")) {
eprintln!("Cannot send message '{message}': {send_error}");
break;
}
}
});
ReceiverStream::new(rx)
}
fn get_intervals() -> impl Stream<Item = u32> {
let (tx, rx) = trpl::channel();
trpl::spawn_task(async move {
let mut count = 0;
loop {
trpl::sleep(Duration::from_millis(1)).await;
count += 1;
if let Err(send_error) = tx.send(count) {
eprintln!("Could not send interval {count}: {send_error}");
break;
};
}
});
ReceiverStream::new(rx)
}
```
*清单 17-40：处理出错并关闭循环*
与往常一样，处理消息发送出错的正确方法各有不同；只要确保咱们有种策略就可以了。
既然我们已经看到了实践种的大量异步，那么我们来退后一步，深入探讨一下 Rust 用于实现异步的 `Future`、`Stream` 及其他关键特质的一些细节。
（End）
**清单 17-30**：成功地将迭代器用作流的基础
将所有这些部分组合在一起后，这段代码就按我们期望的方式运行了！更重要的是，既然现在作用域内有了 `StreamExt`，我们就可以像使用迭代器一样，使用他的所有实用方法。