Rust异步限流库async-throttle使用指南

介绍

async-throttle是一个Rust异步限流库，专门用于管理并发请求和控制请求速率。它特别适合需要限制API调用频率或控制资源消耗的异步应用场景。

主要特性：

• 精确控制请求速率（如每秒/每分钟最大请求数）
• 限制最大并发请求数
• 基于Tokio的异步实现
• 轻量级且易于集成

安装

在Cargo.toml中添加依赖：

[dependencies]
async-throttle = "0.3"
tokio = { version = "1.0", features = ["full"] }

基本使用方法

1. 速率限制（Rate Limiting）

限制每秒最大请求数：

use async_throttle::RateLimiter;
use std::time::Duration;

#[tokio::main]
async fn main() {
    // 限制每秒最多5个请求
    let rate_limiter = RateLimiter::new(5, Duration::from_secs(1));
    
    for i in 0..10 {
        rate_limiter.acquire().await;
        println!("发送请求 {}", i);
        // 这里替换为你的实际异步请求
        // let _ = some_async_request().await;
    }
}

2. 并发限制（Concurrency Limiting）

限制同时进行的最大请求数：

use async_throttle::ConcurrencyLimiter;

#[tokio::main]
async fn main() {
    // 限制最大并发数为3
    let concurrency_limiter = ConcurrencyLimiter::new(3);
    
    let mut handles = vec![];
    
    for i in 0..5 {
        let limiter = concurrency_limiter.clone();
        handles.push(tokio::spawn(async move {
            let _guard = limiter.acquire().await;
            println!("开始处理任务 {}", i);
            tokio::time::sleep(Duration::from_secs(1)).await;
            println!("完成任务 {}", i);
        }));
    }
    
    for handle in handles {
        handle.await.unwrap();
    }
}

高级用法

组合速率和并发限制

use async_throttle::{RateLimiter, ConcurrencyLimiter};

#[tokio::main]
async fn main() {
    // 每秒最多5个请求，同时最多3个并发
    let rate_limiter = RateLimiter::new(5, Duration::from_secs(1));
    let concurrency_limiter = ConcurrencyLimiter::new(3);
    
    for i in 0..10 {
        rate_limiter.acquire().await;
        let _guard = concurrency_limiter.acquire().await;
        
        println!("处理请求 {}", i);
        // 执行你的异步操作
        tokio::time::sleep(Duration::from_millis(200)).await;
    }
}

自定义错误处理

use async_throttle::RateLimiter;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let rate_limiter = RateLimiter::new(5, Duration::from_secs(1));
    
    for i in 0..10 {
        rate_limiter.acquire().await;
        
        match some_fallible_operation(i).await {
            Ok(_) => println!("成功处理 {}", i),
            Err(e) => eprintln!("处理 {} 时出错: {}", i, e),
        }
    }
    
    Ok(())
}

async fn some_fallible_operation(i: i32) -> Result<(), String> {
    if i == 7 {
        Err("模拟错误".to_string())
    } else {
        tokio::time::sleep(Duration::from_millis(100)).await;
        Ok(())
    }
}

实际应用示例

限制API请求

use async_throttle::RateLimiter;
use reqwest;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 根据API限制设置速率（如GitHub API每分钟5000次）
    let rate_limiter = RateLimiter::new(5000, Duration::from_secs(60));
    let client = reqwest::Client::new();
    
    let urls = vec![
        "https://api.github.com/users/octocat",
        "https://api.github.com/users/rust-lang",
    ];
    
    for url in urls {
        rate_limiter.acquire().await;
        
        let response = client.get(url)
            .header("User-Agent", "MyApp")
            .send()
            .await?;
            
        println!("{} 状态: {}", url, response.status());
    }
    
    Ok(())
}

完整示例代码

下面是一个结合速率限制和并发限制的完整示例，模拟API请求场景：

use async_throttle::{RateLimiter, ConcurrencyLimiter};
use reqwest;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 配置速率限制：每分钟最多100个请求
    let rate_limiter = RateLimiter::new(100, Duration::from_secs(60));
    
    // 配置并发限制：最多同时10个请求
    let concurrency_limiter = ConcurrencyLimiter::new(10);
    
    // 模拟API端点列表
    let api_endpoints = (1..=150).map(|i| format!("https://api.example.com/v1/users/{}", i)).collect::<Vec<_>>();
    
    let client = reqwest::Client::new();
    let mut handles = vec![];
    
    for (i, endpoint) in api_endpoints.into_iter().enumerate() {
        let client = client.clone();
        let rate_limiter = rate_limiter.clone();
        let concurrency_limiter = concurrency_limiter.clone();
        
        handles.push(tokio::spawn(async move {
            // 等待速率限制许可
            rate_limiter.acquire().await;
            
            // 等待并发限制许可
            let _guard = concurrency_limiter.acquire().await;
            
            println!("[请求 {}] 开始处理: {}", i, endpoint);
            
            match client.get(&endpoint)
                .header("User-Agent", "AsyncThrottleDemo")
                .send()
                .await 
            {
                Ok(response) => {
                    println!("[请求 {}] 成功 - 状态: {}", i, response.status());
                    // 处理响应数据...
                }
                Err(e) => {
                    eprintln!("[请求 {}] 失败: {}", i, e);
                }
            }
        }));
    }
    
    // 等待所有任务完成
    for handle in handles {
        handle.await?;
    }
    
    Ok(())
}

注意事项

1. RateLimiter和ConcurrencyLimiter都是线程安全的，可以安全地跨线程共享
2. 对于长期运行的应用，考虑使用RateLimiter::builder()进行更精细的配置
3. 在高并发场景下，限流器本身可能成为瓶颈，需要进行性能测试

async-throttle提供了简单而强大的方式来管理异步应用的请求速率和并发量，是构建健壮、API友好的Rust应用的理想选择。