异步调用Page Speed Insights API超限额，如何实现限流？

1. 为什么现有限流逻辑没起作用？ #

你的代码里的限流逻辑完全没控制住实际请求的发起速度，核心问题有两个：

• 任务创建即调度执行：asyncio.ensure_future会把协程立刻加入事件循环等待执行，你在循环里创建任务后，这些任务会马上开始发起API请求，而你加的await asyncio.sleep只是暂停了创建任务的过程，不是暂停请求的发起。哪怕你在创建任务中间sleep，之前创建的任务已经在并发跑了，请求早就发出去了。
• 限流逻辑计算错误：API限制是100秒最多400次，也就是每秒最多4次请求。你设置rate_limit=2（每秒2次），然后判断count >= rate_limit*2才sleep，相当于每创建4个任务才sleep一次，但这个sleep的时间是interval=0.5秒（1/2），等于你每0.5秒就创建4个任务，算下来每秒8次请求，远远超过了API的每秒4次限制，100秒就是800次，和你说的775次接近，自然触发配额超限。

2. 有效实现限流的方案 #

要控制异步请求的速率，需要同时控制并发数和单位时间内的请求总量，这里给你两种可行的实现方式：

方案一：信号量+固定时间窗口限流 #

用asyncio.Semaphore控制同时发起的请求数，再结合时间窗口统计请求次数，超过限制就等待。

import asyncio
import time

async def retrieve_cwv_data(urls_list):    
    site_id = 10234
    # API限制：100秒最多400次 → 每秒最多4次
    max_requests_per_100s = 400
    window_seconds = 100
    # 用信号量控制并发数，避免同时发起太多请求
    semaphore = asyncio.Semaphore(4)
    request_timestamps = []

    async def limited_build_cwv_data(*args):
        async with semaphore:
            # 清理窗口外的时间戳
            now = time.monotonic()
            while request_timestamps and now - request_timestamps[0] > window_seconds:
                request_timestamps.pop(0)
            # 如果当前窗口内请求数已达上限，等待到窗口刷新
            while len(request_timestamps) >= max_requests_per_100s:
                wait_time = window_seconds - (now - request_timestamps[0])
                await asyncio.sleep(wait_time)
                now = time.monotonic()
                # 再次清理过期时间戳
                while request_timestamps and now - request_timestamps[0] > window_seconds:
                    request_timestamps.pop(0)
            # 记录请求时间
            request_timestamps.append(now)
            # 执行API调用
            return await build_cwv_data(*args)

    # 创建所有限流后的任务
    tasks = []
    for url in urls_list:
        tasks.append(limited_build_cwv_data(site_id, url, 'mobile', psi_key))
        tasks.append(limited_build_cwv_data(site_id, url, 'desktop', psi_key))
    
    # 等待所有任务完成
    results = await asyncio.gather(*tasks)
    return results

方案二：令牌桶算法（平滑限流） #

令牌桶算法可以让请求速率更平滑，避免在窗口起始瞬间集中发起请求。

import asyncio
import time

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = capacity  # 令牌桶最大容量
        self.refill_rate = refill_rate  # 每秒补充的令牌数
        self.tokens = capacity
        self.last_refill_time = time.monotonic()

    async def acquire(self):
        now = time.monotonic()
        # 补充令牌
        self.tokens += (now - self.last_refill_time) * self.refill_rate
        if self.tokens > self.capacity:
            self.tokens = self.capacity
        self.last_refill_time = now
        # 如果令牌不足，等待直到有足够令牌
        while self.tokens < 1:
            await asyncio.sleep(0.1)
            now = time.monotonic()
            self.tokens += (now - self.last_refill_time) * self.refill_rate
            self.last_refill_time = now
        self.tokens -= 1

async def retrieve_cwv_data(urls_list):    
    site_id = 10234
    # 100秒400次 → 每秒4次，令牌桶容量设为4，每秒补充4个令牌
    token_bucket = TokenBucket(capacity=4, refill_rate=4)
    semaphore = asyncio.Semaphore(4)  # 控制并发数

    async def limited_build_cwv_data(*args):
        await token_bucket.acquire()
        async with semaphore:
            return await build_cwv_data(*args)

    tasks = []
    for url in urls_list:
        tasks.append(limited_build_cwv_data(site_id, url, 'mobile', psi_key))
        tasks.append(limited_build_cwv_data(site_id, url, 'desktop', psi_key))
    
    results = await asyncio.gather(*tasks)
    return results

关键注意事项 #

• 不要用asyncio.ensure_future直接创建任务后不管，必须在任务执行前加入限流逻辑，确保请求发起前先通过限流校验。
• 结合Semaphore控制并发数，避免同时发起太多请求导致服务器端的额外限制，也能让你的限流逻辑更稳定。
• 优先使用令牌桶算法，它比固定窗口限流更平滑，不容易出现窗口边界的请求突增问题。

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