如何在Python中实现带延迟控制的异步请求？

你的核心问题是同一API实例下的请求没有按10秒间隔串行执行，原因在于代码里的几个关键错误：

• 每次调用async_request都新建asyncio.Semaphore()，局部锁无法限制同一实例的多个请求；
• 等待时间的计算逻辑存在时间差问题；
• 频繁创建aiohttp.ClientSession不符合最佳实践，也可能引发意外问题。

以下是修正后的代码，完全满足你“同一API串行、不同API并发”的需求：

import asyncio
import datetime
import aiohttp
import math

class API1():
    def __init__(self):
        self.api_limit_in_seconds = 10
        self.headers = {...}
        # 初始化上次请求时间为10秒前，允许立即发送第一个请求
        self.last_request_sent_time = datetime.datetime.now() - datetime.timedelta(seconds=self.api_limit_in_seconds)
        self.urls = ["url1", "url2", ...]
        # 每个API实例专属的信号量，限制同一实例下请求串行执行
        self.semaphore = asyncio.Semaphore(1)
        self.session = None

    # 异步上下文管理器，用于创建/关闭ClientSession
    async def __aenter__(self):
        self.session = aiohttp.ClientSession(headers=self.headers)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()

    async def async_request(self, url, type_of_request):
        async with self.semaphore:
            # 精准计算需要等待的时长
            now = datetime.datetime.now()
            elapsed = (now - self.last_request_sent_time).total_seconds()
            wait_time = max(0, self.api_limit_in_seconds - elapsed)
            
            if wait_time > 0:
                await asyncio.sleep(wait_time)
            
            # 发送请求并更新时间
            async with self.session.request(type_of_request, url) as r:
                result = await r.json()
                self.last_request_sent_time = datetime.datetime.now()
                return result

    async def post_requests(self):
        # 并发启动协程，但同一API实例下会被信号量限制为串行
        return await asyncio.gather(*(self.async_request(url, "GET") for url in self.urls))

# 使用示例
async def main():
    async with API1() as api:
        results = await api.post_requests()
        print(results)

if __name__ == "__main__":
    asyncio.run(main())

关键改动说明 #

1. 全局信号量控制串行：把Semaphore改为API实例的属性，同一实例下的所有请求都会争抢这把锁，确保同一API的请求严格按顺序执行。
2. 复用ClientSession：通过异步上下文管理器创建一次会话，避免每次请求新建会话的开销，同时符合aiohttp的官方推荐用法。
3. 精准计算等待时间：先获取当前时间计算时间差，再确定等待时长，避免计算逻辑和sleep执行之间的时间偏差，保证间隔严格≥10秒。
4. 多API实例并发互不干扰：不同的API实例（如API1、API2）拥有各自的信号量和时间记录，它们之间的请求可以同时进行，最大化并发效率。

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