如何将嵌套异步流转为目标流并实现Rust Peers并发连接

我需要处理一个返回Stream<Item = Future<Output = Result<AsyncRead>>>的request_peers()函数，目标是并发连接所有Peer，忽略连接失败的节点，最终返回一个包含成功连接的Stream<Item = AsyncRead>，同时避免将结果转换为Vec。

之前尝试用filter_map串行处理每个Future，导致10个Peer需要10秒才能完成连接；改用buffer_unordered配合collect转Vec可以实现并发，但希望直接基于Stream输出结果，不需要中间的Vec转换。

解决方案 #

直接通过buffer_unordered实现并发处理，同时保持Stream链式调用，无需转换为Vec：

use core::pin::Pin;
use core::task::{Context, Poll};
use std::error::Error;
use tokio::io::{AsyncRead, ReadBuf};
use futures::{Stream, StreamExt};

struct Dummy;

impl AsyncRead for Dummy {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<tokio::io::Result<()>> {
        Poll::Pending
    }
}

fn request_peers() -> impl Stream<Item = impl futures::Future<Output = tokio::io::Result<impl AsyncRead>> + Send + 'static> {
   futures::stream::iter((0..10).map(move |i| {
        println!("实例化Peer {}", i);
        futures::future::ok(Dummy{})
    }))
}

// 直接返回Stream，无需转换为Vec
fn connect(
    peers: impl Stream<Item = impl futures::Future<Output = tokio::io::Result<impl AsyncRead>> + Send + 'static>
) -> impl Stream<Item = impl AsyncRead> {
    peers
        // 包装连接逻辑：等待Future完成，失败则打印错误并转为Option
        .map(|peer_fut| async move {
            match peer_fut.await {
                Ok(peer) => {
                    tokio::time::sleep(core::time::Duration::from_secs(1)).await;
                    println!("连接成功");
                    Some(peer)
                }
                Err(e) => {
                    eprintln!("连接失败: {}", e);
                    None
                }
            }
        })
        // 并发处理任务，最多同时运行50个（None表示无限制并发）
        .buffer_unordered(50)
        // 过滤掉None值，只保留成功的连接实例
        .filter_map(|opt_peer| async move { opt_peer })
}

#[tokio::main]
async fn main() {
    let peers = request_peers();
    let connected_peers = connect(peers);
    
    connected_peers.for_each_concurrent(None, |_peer| async {
        println!("已处理连接");
    }).await;
}

关键实现要点 #

• buffer_unordered(N)：这个方法会并发执行流中的所有Future，参数N指定最大并发数（None表示不限制），替代了串行的filter_map，让所有连接任务同时启动，1秒内就能完成所有连接。
• 链式Stream处理：通过map包装连接逻辑、buffer_unordered并发执行、filter_map过滤失败结果，整个流程始终保持Stream类型，不需要转换为中间集合。
• 类型约束：需要给Future加上Send + 'static约束，因为buffer_unordered要求任务可以跨线程调度（适配tokio多线程运行时）。

内容的提问来源于stack exchange，提问作者Kevin