Rust实现C# Parallel.ForEach等价方法遇编译错误求助

需求与问题 #

作为Rust新手，想要实现类似C# Parallel.ForEach的异步并行执行功能，编写代码后遇到编译错误，涉及dyn Future<Output = ()>的Sized和Sendtrait约束问题。

C# 参考代码 #

Parallel.ForEach(collection, x => {
  // 处理x的逻辑
})

编写的Rust代码 #

pub async fn parallel_run<T>(collection: Vec<T>, callback:  fn(item: T) -> dyn Future<Output= ()>)
{
  for item in collection  
  {       
    tokio::spawn(callback(item)).await.expect("TODO: panic message");
  }
}

编译错误（翻译后） #

类型dyn Future<Output = ()>的值大小在编译时无法确定 [E0277]
编译时未知大小 提示：dyn Future<Output = ()>未实现trait Sized
注意：该约束由tokio::spawn要求
dyn Future<Output = ()>无法安全地在线程间传递 [E0277]
提示：dyn Future<Output = ()>未实现trait Send
注意：该约束由tokio::spawn要求

问题原因 #

1. Sized约束缺失：dyn Future是动态分发的trait对象，编译时无法确定其大小，而tokio::spawn要求传入的future必须是Sized类型（需要在内存中明确分配空间）。
2. Send约束缺失：Tokio的多线程调度器要求任务能在线程间安全传递，dyn Future默认不附带Send约束，无法满足tokio::spawn的要求。
3. 串行执行问题：原代码在循环中直接await每个任务，会导致任务逐个执行，失去并行效果。

正确实现方式 #

方式一：泛型约束（推荐，性能最优） #

使用泛型明确回调函数和返回future的类型，让编译器推断具体实现，避免动态分发开销：

use std::future::Future;
use tokio;

pub async fn parallel_run<T, F, Fut>(collection: Vec<T>, callback: F)
where
    // 回调函数接收T类型参数，返回Fut类型的future
    F: Fn(T) -> Fut,
    // future必须输出空值，且能在线程间传递，生命周期足够长
    Fut: Future<Output = ()> + Send + 'static,
    // 集合元素必须能在线程间传递，生命周期足够长
    T: Send + 'static,
{
    let mut task_handles = Vec::new();

    // 先批量创建所有并行任务
    for item in collection {
        let handle = tokio::spawn(callback(item));
        task_handles.push(handle);
    }

    // 等待所有任务完成
    for handle in task_handles {
        handle.await.expect("并行任务执行失败");
    }
}

方式二：动态分发（适合多类型future场景） #

如果需要回调返回不同类型的future，可以用Box将future装箱，附带Send + 'static约束：

use std::future::Future;
use tokio;

pub async fn parallel_run<T>(collection: Vec<T>, callback: fn(T) -> Box<dyn Future<Output = ()> + Send + 'static>)
where
    T: Send + 'static,
{
    let mut task_handles = Vec::new();

    for item in collection {
        let handle = tokio::spawn(callback(item));
        task_handles.push(handle);
    }

    for handle in task_handles {
        handle.await.expect("并行任务执行失败");
    }
}

说明 #

• 泛型版本性能更优，因为编译器可以做静态分发优化，适合大多数场景。
• 动态分发版本灵活性更高，允许回调返回不同实现的future，但会带来少量运行时开销。
• 确保你的Tokio依赖启用了rt-multi-thread特性（在Cargo.toml中配置tokio = { version = "1.x", features = ["rt-multi-thread", "macros"] }），否则无法支持多线程并行任务。

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