使用Rust新async/await语法批量发送分组HTTP请求并控制并发数

嘿，我完全懂你现在的困扰——用async/await处理这类分组请求还得控制并发量，一开始确实容易摸不着头脑，尤其是不想用一堆组合子把代码搞得晦涩难懂的时候。咱们一步步来解决这个问题：

核心思路梳理 #

你的原始代码是串行执行的：每一组请求都要等前一组完全结束才会开始。要实现并行，我们需要把每组请求封装成独立的异步任务，同时用工具控制同时运行的任务数量（也就是你说的“工作线程”，不过在异步世界里更准确的说法是“并发任务数”）。

Rust的async/await本身只是语法糖，还需要搭配异步运行时（比如tokio或async-std）才能执行异步任务，这一点在nightly版里也是一样的。下面我用最常用的tokio来演示解决方案。

步骤1：封装分组请求逻辑 #

首先把你的每组请求逻辑抽成一个独立的async函数，这样代码更清晰，也方便后续批量处理：

use reqwest;

async fn process_group(client: &reqwest::Client) -> Result<(), reqwest::Error> {
    // 第一组请求
    let response = client.get("https://somesite.com").send().await?.text().await?;
    
    if response.contains("some stuff") {
        // 第二组请求（仅当第一组满足条件时执行）
        let nested_response = client.get("https://somesite.com/something").send().await?.text().await?;
        
        if nested_response.contains("some new stuff") {
            static mut REQ_COUNT: u32 = 0;
            unsafe {
                REQ_COUNT += 1;
                println!("Got response {}", REQ_COUNT);
            }
        }
    }

    Ok(())
}

步骤2：控制并发数的两种方法 #

方法一：用信号量（Semaphore）手动控制 #

如果你需要更精细的控制，tokio提供的Semaphore是个好选择，它能限制同时运行的任务数量，类似线程池的“线程数”控制：

use tokio::sync::Semaphore;
use std::sync::Arc;

#[tokio::main]
async fn main() -> Result<(), reqwest::Error> {
    // 初始化请求客户端（reqwest的Client是可安全克隆的）
    let client = reqwest::Client::new();
    // 设置最大并发数为5，可根据你的需求调整
    let semaphore = Arc::new(Semaphore::new(5));
    // 假设我们要并行处理10组请求
    let total_groups = 10;

    // 存储所有任务的句柄，等待它们完成
    let mut task_handles = Vec::with_capacity(total_groups);

    for _ in 0..total_groups {
        let client_clone = client.clone();
        let semaphore_clone = Arc::clone(&semaphore);

        // 生成异步任务
        let handle = tokio::spawn(async move {
            // 获取信号量许可，没有许可时会等待
            let _permit = semaphore_clone.acquire().await.unwrap();
            // 执行分组请求逻辑
            process_group(&client_clone).await
        });

        task_handles.push(handle);
    }

    // 等待所有任务完成，并处理可能的错误
    for handle in task_handles {
        handle.await.unwrap()?;
    }

    Ok(())
}

原理说明：信号量会维护一个许可池，每次任务开始前必须获取一个许可，任务结束后许可自动释放（_permit离开作用域时会被Drop），这样就保证最多只有5个任务同时执行。

方法二：用futures的buffer_unordered简化代码 #

如果你不需要太精细的控制，futures库的buffer_unordered能更简洁地实现并发控制，代码可读性更高：

use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() -> Result<(), reqwest::Error> {
    let client = reqwest::Client::new();
    let total_groups = 10;
    let max_concurrent = 5;

    // 创建任务流，自动控制并发数
    let task_stream = stream::iter(0..total_groups)
        .map(|_| process_group(&client))
        .buffer_unordered(max_concurrent);

    // 遍历所有任务结果，处理成功/失败
    task_stream.for_each(|result| async {
        match result {
            Ok(_) => {},
            Err(e) => eprintln!("Group request failed: {}", e),
        }
    }).await;

    Ok(())
}

原理说明：buffer_unordered会自动管理任务队列，最多同时运行max_concurrent个任务，当一个任务完成后，会自动启动下一个任务，无需手动处理信号量。

针对你的场景的额外说明 #

1. Rust Nightly兼容性：以上代码完全兼容nightly版的async/await语法，因为tokio和futures都对最新的异步语法提供了完善支持，和stable版的差异极小。
2. 避免组合子困惑：通过把分组逻辑封装成独立的async函数，我们完全避开了复杂的组合子链式调用，代码结构和同步代码几乎一致，可读性大大提升。
3. 性能注意点：reqwest::Client是线程安全且可克隆的，不要在每个任务里新建Client，克隆已有的实例能复用连接池，提升性能。

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