如何在Rust中实现批量Future竞速并支持重试

你可以通过futures crate的select_all方法解决批量Future竞速的问题，再结合循环+任务生成闭包实现无效结果的重试逻辑，具体方案如下：

核心依赖 #

首先在Cargo.toml中添加必要依赖：

tokio = { version = "1.0", features = ["full"] }
futures = "0.3"
rand = "0.8" # 仅用于示例模拟随机结果，实际场景可移除

完整实现示例 #

1. 模拟带无效结果的异步任务 #

先定义一个模拟的异步任务，它会随机返回有效/无效结果：

use rand::Rng;
use tokio::time::{sleep, Duration};

// 模拟异步任务：返回正数为有效，负数为无效
async fn fetch_data(task_id: u32) -> i32 {
    // 模拟网络延迟
    sleep(Duration::from_millis(rand::thread_rng().gen_range(100..500))).await;
    
    if rand::thread_rng().gen() {
        println!("[任务{}] 返回有效结果", task_id);
        task_id as i32
    } else {
        println!("[任务{}] 返回无效结果", task_id);
        -(task_id as i32)
    }
}

2. 实现竞速+重试逻辑 #

使用select_all批量竞速Future，同时支持无效结果的重试（可配置最大重试次数）：

use futures::future::{select_all, BoxFuture};
use std::pin::Pin;

// 封装可重试的任务：包含任务ID、生成器、最大重试次数、已重试次数
struct RetryableTask<T> {
    task_id: u32,
    task_factory: Box<dyn Fn() -> Pin<Box<dyn Future<Output = T>>>>,
    max_retries: usize,
    retries_used: usize,
}

async fn race_with_retry<T>(
    mut tasks: Vec<RetryableTask<T>>,
    is_valid: impl Fn(&T) -> bool,
) -> Option<T> {
    // 初始化所有任务的Future实例
    let mut futures: Vec<_> = tasks
        .iter()
        .map(|task| ((task.task_factory)(), task))
        .collect();

    while !futures.is_empty() {
        // 竞速所有当前运行的Future，返回首个完成的结果、对应任务、剩余未完成的Future
        let ((result, task), _, remaining_futures) = select_all(futures).await;

        // 检查结果是否有效
        if is_valid(&result) {
            // 找到有效结果，直接返回；剩余Future会被自动drop，Tokio会终止对应的异步任务
            return Some(result);
        } else {
            // 检查是否还有重试次数
            if task.retries_used < task.max_retries {
                println!("[任务{}] 重试中（已用{}次，剩余{}次）", task.task_id, task.retries_used + 1, task.max_retries - task.retries_used - 1);
                // 创建新的重试任务实例，更新重试计数
                let new_task = RetryableTask {
                    task_id: task.task_id,
                    task_factory: task.task_factory.clone(),
                    max_retries: task.max_retries,
                    retries_used: task.retries_used + 1,
                };
                // 生成重试任务的Future并加入剩余队列
                let mut updated_futures = remaining_futures;
                updated_futures.push(((new_task.task_factory)(), &new_task));
                futures = updated_futures;
            } else {
                println!("[任务{}] 已耗尽所有重试次数，放弃", task.task_id);
                // 无重试次数，直接使用剩余Future继续竞速
                futures = remaining_futures;
            }
        }
    }

    // 所有任务重试后仍无有效结果
    None
}

3. 调用示例 #

在主函数中创建任务并执行竞速逻辑：

#[tokio::main]
async fn main() {
    // 创建5个可重试任务，每个最多重试2次
    let tasks = vec![
        RetryableTask {
            task_id: 1,
            task_factory: Box::new(|| Box::pin(fetch_data(1))),
            max_retries: 2,
            retries_used: 0,
        },
        RetryableTask {
            task_id: 2,
            task_factory: Box::new(|| Box::pin(fetch_data(2))),
            max_retries: 2,
            retries_used: 0,
        },
        RetryableTask {
            task_id: 3,
            task_factory: Box::new(|| Box::pin(fetch_data(3))),
            max_retries: 2,
            retries_used: 0,
        },
        RetryableTask {
            task_id: 4,
            task_factory: Box::new(|| Box::pin(fetch_data(4))),
            max_retries: 2,
            retries_used: 0,
        },
        RetryableTask {
            task_id: 5,
            task_factory: Box::new(|| Box::pin(fetch_data(5))),
            max_retries: 2,
            retries_used: 0,
        },
    ];

    // 定义有效结果的判断规则：正数为有效
    let is_valid = |val: &i32| *val > 0;

    match race_with_retry(tasks, is_valid).await {
        Some(valid_result) => println!("最终获取有效结果：{}", valid_result),
        None => println!("所有任务重试后仍无有效结果"),
    }
}

关键逻辑说明 #

1. 批量竞速：futures::select_all接收一个Future集合，自动竞速并返回首个完成的结果，解决了tokio::select!需要显式列出所有Future的限制。
2. 重试机制：通过任务生成闭包（task_factory）可以重新创建任务的Future实例，当结果无效且还有重试次数时，将新的Future加入剩余队列继续竞速。
3. 自动终止未完成任务：未完成的Future在被drop时，Tokio会自动终止对应的异步任务，无需手动处理取消逻辑。

注意事项 #

• 任务生成闭包必须能够重新创建相同的任务（比如HTTP请求需要重新发起，而不是复用原请求实例）。
• 如果不需要重试，只需移除重试逻辑，直接在结果无效时丢弃该任务，继续竞速剩余Future即可。
• 可以根据需求调整重试策略，比如全局重试次数限制、指数退避重试等。

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