如何在rayon并行迭代器的map中传入返回Result的异步闭包

问题背景 #

现有两个异步函数add_two_num和check_num，核心逻辑是在异步函数中调用不同返回类型的anyhow::Result<T>：

use anyhow::{Result, Ok};
use rayon::prelude::*;

pub async fn add_two_num(num_one: u64, num_two: u64) -> Result<u64> {
    check_num(num_one).await?;
    check_num(num_two).await?;
    anyhow::Ok(num_one + num_two)
}

pub async fn check_num(num: u64) -> Result<()> {
    assert!(num <= u64::MAX);
    Ok(())
}

尝试在主函数中用Rayon将两个Vec转为并行迭代器并zip，在map闭包内调用异步函数时遇到两个问题：

1. 异步闭包无法直接传入rayon::IntoParallelIterator.map；
2. 闭包内使用?运算符时，无法正确标注返回anyhow::Result<T>类型。

需求是：

• 将两个并行迭代器zip后传入闭包；
• 闭包内调用异步函数，通过?提前返回错误。

解决方案 #

问题1：适配Rayon同步迭代器与异步函数 #

Rayon的并行迭代器是同步执行的，map方法仅接收同步闭包，不能直接传入异步闭包。解决方式是在闭包内部利用Tokio runtime的句柄，将异步任务的执行转为同步阻塞：

• 先获取当前Tokio runtime的句柄Handle::current()；
• 在并行闭包中调用handle.block_on()，将异步函数的执行包裹为同步操作。

问题2：闭包的Result返回类型处理 #

闭包需要返回anyhow::Result<u64>类型，这样?运算符才能正常传递错误。可以通过显式标注变量类型（如Vec<Result<u64>>）让编译器自动推断闭包返回值，也可以在闭包中显式标注返回类型。

完整修正代码 #

use anyhow::{Result, Ok};
use rayon::prelude::*;
use tokio::runtime::Handle;

pub async fn add_two_num(num_one: u64, num_two: u64) -> Result<u64> {
    check_num(num_one).await?;
    check_num(num_two).await?;
    anyhow::Ok(num_one + num_two)
}

pub async fn check_num(num: u64) -> Result<()> {
    assert!(num <= u64::MAX);
    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    let nums_one: Vec<u64> = vec![1, 2, 3, 4, 5, 6, 7];
    let nums_two: Vec<u64> = vec![1, 2, 3, 4, 5, 6, 7];

    // 获取当前Tokio Runtime的句柄，用于在同步闭包中运行异步任务
    let handle = Handle::current();

    // 显式标注结果类型为Vec<Result<u64>>，帮助编译器推断闭包返回值
    let results: Vec<Result<u64>> = nums_one.into_par_iter()
        .zip_eq(nums_two.into_par_iter())
        .map(move |(num_one, num_two)| {
            // 用handle.block_on将异步执行转为同步，适配Rayon的并行迭代器
            handle.block_on(async move {
                add_two_num(num_one, num_two).await
            })
        })
        .collect();

    // 示例：遍历处理所有结果，打印成功值或错误信息
    for res in results {
        match res {
            Ok(sum) => println!("计算结果：{}", sum),
            Err(e) => eprintln!("发生错误：{}", e),
        }
    }

    Ok(())
}

关键说明 #

• 使用Handle::current()复用当前Tokio Runtime，避免在每个并行任务中重复创建Runtime，提升性能；
• handle.block_on()将异步函数的执行阻塞到完成，将异步结果转为同步返回，满足Rayon并行迭代器的要求；
• 闭包返回的Result<u64>自动承接add_two_num的返回值，内部的?运算符已完成错误传递，无需额外处理。

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