Rust 异步流实战：用 futures::stream 做并发拉取与背压

为什么用 Stream

一次拉一个 future 用 await 足够；要处理“一串异步产生的值”（分页抓取、消息订阅、文件分块）时，Stream 比手动维护 Vec 干净得多。futures crate 的 StreamExt 提供了 map/filter/for_each_concurrent 等组合子，把背压和并发控制一次解决。

一、把一个迭代器变成流

用 stream::iter 把普通集合包成 Stream，再逐个异步处理。

use futures::stream::{self, StreamExt};

let urls = vec!["/a", "/b", "/c"];
let mut s = stream::iter(urls).map(|u| async move {
    format!("fetched {u}")
});

while let Some(v) = s.next().await {
    println!("{v}");
}

二、并发拉取并限制并行度

for_each_concurrent 配合 buffer_unordered 限制同时最多 4 个在途请求，天然形成背压，不会因为目标很多就瞬间打满连接。

use futures::stream::{self, StreamExt};

let tasks = 0..100;
stream::iter(tasks)
    .map(|i| async move {
        reqwest::get(format!("https://example.com/{i}")).await
    })
    .buffer_unordered(4) // 最多 4 个在途
    .for_each_concurrent(4, |resp| async move {
        if let Ok(r) = resp {
            println!("status {}", r.status());
        }
    })
    .await;

三、用 fold 做流式聚合

不想 collect 到内存，可以用 fold 边流边累计，处理大量数据时内存占用稳定。

use futures::stream::{self, StreamExt};

let total: usize = stream::iter(1..=1000)
    .map(|n| async move { n })
    .fold(0usize, |acc, n| async move { acc + n })
    .await;

四、自己实现异步流

需要自定义产出节奏时，实现 Stream trait 的 poll_next，决定什么时候给下一个值、什么时候 Pending。

use std::pin::Pin;
use std::task::{Context, Poll};
use futures::stream::Stream;

struct Counter { n: u32, max: u32 }

impl Stream for Counter {
    type Item = u32;
    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<u32>> {
        if self.n >= self.max {
            Poll::Ready(None)
        } else {
            let v = self.n;
            self.n += 1;
            Poll::Ready(Some(v))
        }
    }
}

小结

Stream 把“一串异步值”的表达标准化：buffer_unordered 控并发、fold 做流式聚合、自己实现 poll_next 定节奏。比起手动管理一堆 JoinHandle，背压和取消都能借由流的撤销一次性处理干净。