Python 3.8异步请求控速：每秒15次调用及重试实现问询

我用Python 3.8写了个异步函数analyse_doc，用来封装Web服务的HTTP请求，现在要处理约1000份文档，要求尽可能快完成。这个服务有严格的限流规则：每秒最多15次事务（不是任意时刻15个并发请求），每秒调用超过15次会触发429、503或504错误。

我想知道能不能在Python里实现这样的逻辑：每秒异步发送15个请求，等1秒后再重复操作直到任务队列清空，同时支持失败任务的重试。

目前我的代码是无界并行模式（没用到信号量），但已经实现了重试逻辑，代码如下：

tasks = {asyncio.create_task(analyse_async(doc)): doc for doc in documents}
pending = set(tasks)

# Handle retry
while pending:
    #  backoff in case of 429
    time.sleep(1)

    # concurrent call return_when all completed
    finished, pending = await asyncio.wait(
        pending, return_when=asyncio.ALL_COMPLETED
    )

    
    # check  if task has exception and register for new run.
    for task in finished:
        arg = tasks[task]

        if task.exception():
            new_task = asyncio.create_task(analyze_doc(doc))
            tasks[new_task] = doc
            pending.add(new_task)

原代码存在的问题 #

1. 同步阻塞问题：用time.sleep(1)会卡住整个异步事件循环，完全浪费了异步IO的优势，导致任务执行效率极低。
2. 突破限流：无界并发会瞬间发起大量请求，直接触发服务的限流错误。
3. 变量错误：重试逻辑里的new_task = asyncio.create_task(analyze_doc(doc))存在两个问题：一是函数名和前面的analyse_async不一致，二是用了未定义的doc变量，应该用arg。

正确实现方案 #

要满足每秒15次请求的限流要求，同时保留重试逻辑，我们可以调整为批量调度+异步等待的模式，具体实现如下：

1. 完善带重试的请求函数

先给请求函数加上重试逻辑，针对限流错误做退避重试：

import asyncio

async def analyse_async(doc, max_retries=3):
    retries = 0
    while retries < max_retries:
        try:
            # 替换成你实际的analyse_doc调用逻辑
            return await analyse_doc(doc)
        except Exception as e:
            # 根据你使用的HTTP库，判断是否是限流相关错误（比如aiohttp的ClientResponseError）
            # 示例：假设e.status是429/503/504时触发重试
            if hasattr(e, 'status') and e.status in (429, 503, 504):
                retries += 1
                # 指数退避，避免再次触发限流
                await asyncio.sleep(2 ** retries)
            else:
                # 非限流错误，直接抛出
                raise
    # 超过最大重试次数，抛出异常
    raise RuntimeError(f"文档处理失败，已重试{max_retries}次: {doc}")

2. 实现每秒批量调度逻辑

把文档按每15个一组拆分，每秒调度一组，同时等待所有任务完成：

async def process_documents(documents):
    # 将文档分成每15个一组，确保每秒最多发送15次请求
    doc_batches = [documents[i:i+15] for i in range(0, len(documents), 15)]
    all_tasks = []

    for batch in doc_batches:
        # 为当前批量创建异步任务
        batch_tasks = [asyncio.create_task(analyse_async(doc)) for doc in batch]
        all_tasks.extend(batch_tasks)
        # 异步等待1秒，不阻塞事件循环，让已发起的请求正常处理
        await asyncio.sleep(1)
    
    # 等待所有任务完成，return_exceptions=True会收集异常而不是终止程序
    results = await asyncio.gather(*all_tasks, return_exceptions=True)
    
    # 可选：处理失败的任务结果
    for idx, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"文档{documents[idx]}处理失败: {str(result)}")

# 运行事件循环
asyncio.run(process_documents(documents))

方案说明 #

• 严格限流：通过批量拆分+异步sleep，确保每秒最多发起15次请求，完全符合服务的限流规则。
• 异步友好：用asyncio.sleep()替代同步sleep，不会阻塞事件循环，已发起的请求和重试任务可以在等待下一批的时间里正常执行。
• 灵活重试：针对限流错误做指数退避，避免频繁触发服务拦截，同时可以通过max_retries控制重试次数。
• 错误处理：最后用asyncio.gather收集所有结果，方便统一处理失败任务。

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