如何使用Rust实现并行请求并在执行完成后收集返回结果

实现方案 #

你要的并行请求+保留返回结果的Rust实现如下，正常情况下相比串行15秒的耗时可以压缩到2-3秒左右（取决于接口响应速度和并发限制）：

全并发实现（适合接口无严格频率限制的场景） #

首先在Cargo.toml中引入必要依赖：

[dependencies]
tokio = { version = "1.0", features = ["full"] }
futures = "0.3"

业务代码实现：

use futures::future::join_all;

// 先批量生成所有请求的Future，不执行
let mut requests = Vec::with_capacity((r.total_pages - 2) as usize);
for page in 2..r.total_pages {
    requests.push(get(page));
}

// 并行等待所有请求返回，结果顺序和请求顺序完全一致
let responses = join_all(requests).await;

// 合并所有返回结果
for resp in responses {
    // 可根据需求补充错误处理逻辑
    let r = resp.expect("请求失败");
    auctions = parse_hypixel(r.auctions, auctions);
}

限流并发实现（适合接口有频率限制的场景） #

如果全并发触发了接口限流导致速度反而变慢，可以通过限制最大并发数优化：

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

// 最大并发数可根据接口实际限制调整，比如10、20
const MAX_CONCURRENT: usize = 10;
let mut request_stream = FuturesUnordered::new();

for page in 2..r.total_pages {
    request_stream.push(get(page));
    // 达到并发上限时，先等待一个请求完成再提交新请求
    if request_stream.len() >= MAX_CONCURRENT {
        if let Some(Ok(r)) = request_stream.next().await {
            auctions = parse_hypixel(r.auctions, auctions);
        }
    }
}

// 处理剩余未完成的请求
while let Some(Ok(r)) = request_stream.next().await {
    auctions = parse_hypixel(r.auctions, auctions);
}

之前并行版本速度更慢通常是三个原因导致的：1. 实现时加了不必要的全局锁，导致所有请求还是串行执行；2. 并发数太高触发了服务端的频率限制，接口返回延迟变高甚至报错；3. 每次请求都单独创建线程，线程开销超过了并行带来的收益。上面的实现用异步任务调度替代手动线程管理，同时支持灵活调整并发数，可以解决上述问题。

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