如何控制Rust HTTP请求QPS以测试网站限流功能

要实现每秒恰好发送100次请求并保存响应结果，可以通过结合tokio的定时功能与信号量（Semaphore）精准控制请求速率，同时保留异步请求的优势。以下是具体实现方案：

依赖更新 #

先确保Cargo.toml添加tokio-util依赖（用于信号量控制）：

[dependencies]
reqwest = { version = "0.11.6" }
tokio = { version = "1.14.0", features = ["full"] }
futures = "0.3.24"
tokio-util = { version = "0.7.3", features = ["sync"] }

修改后的代码 #

use futures::stream::StreamExt;
use reqwest::Client;
use tokio::time::{interval, Duration};
use tokio_util::sync::Semaphore;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 配置参数：每秒请求数、总请求数、目标URL
    let requests_per_second = 100;
    let total_requests = 500;
    let target_url = "https://api.ipify.org/";

    // 全局HTTP客户端（复用连接池，避免重复创建）
    let client = Client::new();
    // 用于保存响应结果（此处存储状态码，可按需扩展为响应体等）
    let mut responses = Vec::with_capacity(total_requests);

    // 信号量初始许可为0，后续每秒补充100个
    let semaphore = Semaphore::new(0);
    // 每秒触发一次的定时器
    let mut interval = interval(Duration::from_secs(1));

    // 提前释放第一波许可，避免等待1秒才启动请求
    semaphore.add_permits(requests_per_second);

    // 生成所有请求任务
    let request_stream = futures::stream::iter(0..total_requests).map(|_| {
        let client = client.clone();
        let semaphore = semaphore.clone();
        let url = target_url.to_string();

        async move {
            // 获取许可，等待速率控制
            let _permit = semaphore.acquire().await.unwrap();
            
            // 发送请求并处理结果
            match client.get(&url).send().await {
                Ok(resp) => {
                    let status = resp.status();
                    println!("请求完成，状态码: {}", status);
                    Some(status)
                }
                Err(e) => {
                    eprintln!("请求失败: {}", e);
                    None
                }
            }
        }
    });

    // 后台启动定时器，每秒补充请求许可
    tokio::spawn(async move {
        loop {
            interval.tick().await;
            semaphore.add_permits(requests_per_second);
        }
    });

    // 收集所有响应结果
    responses.extend(request_stream.buffer_unordered(requests_per_second).await);

    println!("所有请求完成，共收到{}个有效响应", responses.iter().flatten().count());
    Ok(())
}

核心实现说明 #

• 信号量控制：通过Semaphore限制并发请求数，结合定时器每秒补充100个许可，确保每秒最多发送100次请求。
• 全局客户端复用：避免为每个请求创建新的Client，复用连接池提升请求效率。
• 响应结果存储：用Vec保存请求状态码，可根据需求修改为存储响应体、响应时间等信息。
• 定时补许可：tokio::time::interval每秒触发一次，向信号量添加许可，维持稳定的请求速率。

替代方案：专业速率限制库 #

如果需要更复杂的速率控制（如令牌桶、突发请求限制），可以使用ratelimit_meter库：

1. 添加依赖：

ratelimit_meter = "0.11.0"

2. 核心控制片段：

use ratelimit_meter::{DirectRateLimiter, LeakyBucket};
use std::time::Duration;

// 创建漏桶限流器：每秒允许100次请求
let limiter = DirectRateLimiter::new(LeakyBucket::new(100, Duration::from_secs(1)));

// 每次请求前等待令牌，确保速率合规
limiter.wait().await.unwrap();
// 后续执行请求逻辑...

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