如何在Rust中并行执行不同返回类型的异步函数？

问题描述 #

在Rust项目的process函数中，调用了两个返回不同类型（A和B）的异步函数func_a与func_b，原本为串行执行，改为用tokio::task::JoinSet并行执行时出现编译错误：

expected {async block@src/service.rs:34:17: 34:69} to be a future that resolves to Box<impl Future<Output = Result<A, Error>>>,
but it resolves to Box<impl Future<Output = Result<B, Error>>> expected struct tokio::task::JoinHandle<Box<impl std::future::Future<Output = Result<A, actix_web::Error>>>> found structtokio::task::JoinHandle<Box<impl std::future::Future<Output = Result<B, actix_web::Error>>>>distinct uses ofimpl Trait` result in different opaque types

问题原因 #

tokio::task::JoinSet要求所有加入集合的任务必须返回完全相同的类型，但func_a返回Result<A, actix_web::Error>，func_b返回Result<B, actix_web::Error>，两者的输出类型不一致，编译器将其判定为不同的不透明类型，因此报错。

解决方案 #

方法1：定义枚举统一返回类型（推荐） #

创建一个枚举来包裹两种不同的结果类型，让两个任务的返回值统一为该枚举的Result类型：

use actix_web::Error;
use tokio::task::JoinSet;

// 定义枚举，覆盖两种可能的返回结果
enum TaskResult {
    A(A),
    B(B),
}

async fn process() -> Result<(A, B), Error> {
    let mut join_set = JoinSet::new();

    // 提交func_a任务，将结果包装为TaskResult::A
    join_set.spawn(async {
        func_a().await.map(TaskResult::A)
    });

    // 提交func_b任务，将结果包装为TaskResult::B
    join_set.spawn(async {
        func_b().await.map(TaskResult::B)
    });

    // 遍历收集结果
    let mut a = None;
    let mut b = None;

    while let Some(task_res) = join_set.join_next().await {
        match task_res? {
            TaskResult::A(val) => a = Some(val),
            TaskResult::B(val) => b = Some(val),
        }
    }

    // 这里假设两个任务都能成功返回，实际可根据需求处理空值情况
    Ok((a.unwrap(), b.unwrap()))
}

方法2：使用动态类型装箱（不推荐） #

如果不想定义枚举，可以将结果装箱为Box<dyn Any + Send>，但这种方式需要手动处理类型转换，存在类型不安全的风险：

use actix_web::Error;
use std::any::Any;
use tokio::task::JoinSet;

async fn process() -> Result<(A, B), Error> {
    let mut join_set = JoinSet::new();

    join_set.spawn(async {
        func_a().await.map(|val| Box::new(val) as Box<dyn Any + Send>)
    });

    join_set.spawn(async {
        func_b().await.map(|val| Box::new(val) as Box<dyn Any + Send>)
    });

    let mut a = None;
    let mut b = None;

    while let Some(task_res) = join_set.join_next().await {
        let boxed_val = task_res?;
        // 尝试转换为A类型
        if let Some(val) = boxed_val.downcast_ref::<A>() {
            a = Some(val.clone()); // 需确保A实现Clone，或用所有权转移方式
        }
        // 尝试转换为B类型
        else if let Some(val) = boxed_val.downcast_ref::<B>() {
            b = Some(val.clone());
        }
    }

    Ok((a.unwrap(), b.unwrap()))
}

总结 #

优先选择枚举统一类型的方案，它类型安全、可读性强，能避免动态类型转换带来的潜在问题。

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