Asyncio如何实现每秒仅发1次API请求，避免返回429错误

代码修改方案 #

你当前的代码会一次性创建所有请求任务并发执行，自然会触发API的频率限制，按以下方式修改即可实现每秒1次的请求控制：

import asyncio
import aiohttp

async def ps_request_marshal(converted_urls, device_input):
    data = []
    # 避免使用全局变量，直接在函数内处理逻辑即可
    url = 'https://www.googleapis.com/pagespeedonline/v5/runPagespeed?url={}&key=+key&strategy=' + device_input

    async with aiohttp.ClientSession() as session:
        for page in converted_urls:
            # 发起单次请求
            response = await session.get(url.format(page), ssl=False)
            data.append(await response.json())
            # 等待1秒后发起下一次请求
            await asyncio.sleep(1)
        print(data)
    # 直接返回结果，不需要用全局变量传递
    return data

如果想要更灵活的速率控制，也可以封装通用的异步限流器，后续调整速率不需要修改业务逻辑：

import asyncio
import time

class RateLimiter:
    def __init__(self, max_per_second):
        self.max_per_second = max_per_second
        self.last_call = 0
        self.lock = asyncio.Lock()

    async def acquire(self):
        async with self.lock:
            now = time.time()
            # 计算需要等待的时间
            wait_time = 1/self.max_per_second - (now - self.last_call)
            if wait_time > 0:
                await asyncio.sleep(wait_time)
            self.last_call = time.time()

# 业务代码使用方式
async def ps_request_marshal(converted_urls, device_input):
    data = []
    url = 'https://www.googleapis.com/pagespeedonline/v5/runPagespeed?url={}&key=+key&strategy=' + device_input
    # 初始化限流器，每秒1次请求
    limiter = RateLimiter(1)

    async with aiohttp.ClientSession() as session:
        for page in converted_urls:
            # 先获取限流许可
            await limiter.acquire()
            response = await session.get(url.format(page), ssl=False)
            data.append(await response.json())
        print(data)
    return data

注意事项 #

• 如果你需要在请求失败时重试，建议在重试逻辑中也添加相同的限流等待，避免重试请求触发频率限制
• 第二种限流器方案会自动补足等待时间，如果单次请求响应超过1秒，下一次请求会直接发起不会额外等待，在符合速率要求的前提下提升效率

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