Python脚本并发请求如何控制速率

本文目录导读：

1. 使用 asyncio + asyncio.Semaphore + asyncio.sleep
2. 使用 asyncio 令牌桶算法
3. 使用第三方库 aiolimiter
4. 使用 ThreadPoolExecutor + rate_limit 库
5. 使用 asyncio.Queue 精细控制
6. 优缺点对比
7. 推荐选择

在Python中控制并发请求的速率通常有几种常用方法,我来介绍最实用的几种方案：

使用 asyncio + asyncio.Semaphore + asyncio.sleep

import asyncio
import aiohttp
import time
class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        """
        max_calls: 在period秒内允许的最大请求数
        period: 时间窗口（秒）
        """
        self.max_calls = max_calls
        self.period = period
        self.sem = asyncio.Semaphore(max_calls)
        self.calls = []
    async def acquire(self):
        async with self.sem:
            now = time.monotonic()
            # 清理过期的调用记录
            self.calls = [t for t in self.calls if now - t < self.period]
            if len(self.calls) >= self.max_calls:
                # 等待直到有位置释放
                wait_time = self.period - (now - self.calls[0])
                if wait_time > 0:
                    await asyncio.sleep(wait_time)
            self.calls.append(time.monotonic())
async def limited_request(session, url, rate_limiter):
    await rate_limiter.acquire()
    async with session.get(url) as response:
        return await response.text()
async def main():
    rate_limiter = RateLimiter(max_calls=5, period=1.0)  # 每秒5个请求
    urls = [...]  # 你的URL列表
    async with aiohttp.ClientSession() as session:
        tasks = [limited_request(session, url, rate_limiter) for url in urls]
        results = await asyncio.gather(*tasks)

使用 asyncio 令牌桶算法

更精确的速率控制可以使用令牌桶算法：

import asyncio
import time
class TokenBucket:
    def __init__(self, rate: float, burst_size: int = None):
        """
        rate: 每秒产生的令牌数（请求数）
        burst_size: 最大令牌桶容量（突发请求数）
        """
        self.rate = rate
        self.burst_size = burst_size or rate
        self.tokens = self.burst_size
        self.last_time = time.monotonic()
        self.lock = asyncio.Lock()
    async def acquire(self, tokens: int = 1):
        while True:
            async with self.lock:
                now = time.monotonic()
                # 计算应添加的令牌
                elapsed = now - self.last_time
                self.tokens = min(
                    self.burst_size,
                    self.tokens + elapsed * self.rate
                )
                self.last_time = now
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return True
            # 没有足够令牌，等待
            await asyncio.sleep(0.01)  # 10ms检查一次
async def main():
    bucket = TokenBucket(rate=10, burst_size=20)  # 每秒10个请求，突发20个
    async with aiohttp.ClientSession() as session:
        for url in urls:
            await bucket.acquire()
            async with session.get(url) as response:
                # 处理响应
                pass

使用第三方库 aiolimiter

import asyncio
import aiohttp
from aiolimiter import AsyncLimiter
async def main():
    # 创建限速器：每秒最多10个请求
    limiter = AsyncLimiter(max_rate=10, time_period=1)
    async with aiohttp.ClientSession() as session:
        async def fetch(url):
            async with limiter:
                async with session.get(url) as response:
                    return await response.text()
        tasks = [fetch(url) for url in urls]
        results = await asyncio.gather(*tasks)
# 安装：pip install aiolimiter

使用 ThreadPoolExecutor + rate_limit 库

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
from ratelimit import limits, sleep_and_retry
@sleep_and_retry
@limits(calls=10, period=1)  # 每秒10次
def fetch_url(url):
    response = requests.get(url)
    return response.text
def main():
    urls = [...]  # URL列表
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(fetch_url, url): url for url in urls}
        for future in as_completed(futures):
            url = futures[future]
            try:
                result = future.result()
                print(f"完成: {url}")
            except Exception as e:
                print(f"失败: {url}, 错误: {e}")
# 安装：pip install ratelimit requests

使用 asyncio.Queue 精细控制

import asyncio
import aiohttp
import time
async def rate_limited_worker(worker_id, queue, session, rate_limit=1.0):
    """rate_limit: 每秒处理的请求数"""
    interval = 1.0 / rate_limit
    last_request_time = 0
    while True:
        try:
            url = await asyncio.wait_for(queue.get(), timeout=1.0)
        except asyncio.TimeoutError:
            break
        # 速率控制
        now = time.monotonic()
        wait_time = interval - (now - last_request_time)
        if wait_time > 0:
            await asyncio.sleep(wait_time)
        try:
            async with session.get(url) as response:
                result = await response.text()
                print(f"Worker {worker_id}: 完成 {url}")
        except Exception as e:
            print(f"Worker {worker_id}: 错误 {url}: {e}")
        last_request_time = time.monotonic()
        queue.task_done()
async def main():
    urls = [...]  # URL列表
    queue = asyncio.Queue()
    for url in urls:
        await queue.put(url)
    async with aiohttp.ClientSession() as session:
        workers = [
            rate_limited_worker(i, queue, session, rate_limit=2.0)  # 每个worker每秒2个请求
            for i in range(5)  # 5个worker
        ]
        await asyncio.gather(*workers)

优缺点对比

推荐选择

• 简单需求：使用 aiolimiter
• 精确控制：令牌桶算法
• 同步场景：ratelimit 库
• 需要自定义调度：Queue模式

使用示例（推荐方案）：

# 最简单的异步速率控制
pip install aiolimiter
from aiolimiter import AsyncLimiter
import asyncio
import aiohttp
async def fetch_with_rate_limit(urls, max_rate=10):
    limiter = AsyncLimiter(max_rate, 1)  # 每秒10个请求
    async with aiohttp.ClientSession() as session:
        async def bounded_fetch(url):
            async with limiter:
                async with session.get(url) as resp:
                    return await resp.text()
        tasks = [bounded_fetch(url) for url in urls]
        return await asyncio.gather(*tasks)