Rust异步流中filter_map无法搭配buffer_unordered并发运行的原因

核心差异来自futures库中StreamExt的map和filter_map两个方法的trait约束完全不同，把给map用的写法直接套给filter_map，就会触发类型链路不匹配的问题。

两个方法的约束细节 #

map 的设计要求 #

map对传入闭包的返回值没有额外限制，只要求闭包能对每个流元素产出一个任意类型的值：

fn map<T, F>(self, f: F) -> Map<Self, F>
where
    F: FnMut(Self::Item) -> T,

之前能正常运行的代码，本质上是利用了这个无约束的特性：

• 传入的async闭包返回值是一个实现了Future的异步任务
• map会原样把每个输入元素替换成对应的异步任务，生成的新流的Item类型就是这些Future本身
• 后续接buffer_unordered(n)时，刚好匹配“流的每个元素都是可并发执行的Future”的要求，所以整条链路能编译通过。

filter_map 的设计要求 #

filter_map从设计上就是为同步的边映射边过滤逻辑准备的，它强制要求传入闭包的返回值必须是Option<T>，不允许返回其他类型：

fn filter_map<T, F>(self, f: F) -> FilterMap<Self, F>
where
    F: FnMut(Self::Item) -> Option<T>,

把async闭包直接塞给filter_map的时候，类型推导会直接出错：

1. 编译器会把async闭包返回的Future，错误当成是Option内部的T类型
2. 推导后得到的流Item类型是这个Future执行完才会产出的整数（比如i32）
3. 后续接buffer_unordered时，它要求流的Item必须实现Future trait，但现在的Item是普通整数，根本没有实现这个trait，就会抛出"{integer} is not a Future"的报错。

正确的并发过滤实现方式 #

要实现异步的边过滤边映射+并发控制，正确的调用顺序是：

• 用map把每个输入元素映射成独立的异步任务（任务内部返回Option）
• 用buffer_unordered控制最大并发数，执行完所有异步任务
• 拿到所有任务的输出（都是Option类型）后，再用filter_map完成同步的过滤和取值
  对应的代码示例：

// 这个写法无需nightly的async_closure特性，stable版本即可运行
use futures::{stream, StreamExt};

async fn foo() {
    let xs_new = stream::once(async { 42 })
        // 普通闭包返回独立的异步任务，任务内部返回Option
        .map(|x| async move {
            if x % 2 == 0 { Some(x + 1) } else { None }
        })
        // 控制最大并发数执行所有异步任务
        .buffer_unordered(42)
        // 对执行结果做同步过滤映射，丢弃None、取出Some值
        .filter_map(|x| x);
}

注意：不要把异步逻辑直接写在filter_map的闭包里，filter_map本身不支持接收返回Future的闭包，这是和map最核心的区别。

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