Python asyncio 失败任务召回与API请求限流管控实现

先明确原代码的核心问题 #

• 手动await asyncio.sleep(1/3)是假限流：仅在创建初始任务时做间隔，任务创建后并发执行；且重试任务完全绕开这个创建流程，直接发请求
• 协程内部死循环重试（try_make_request）：重试时不会校验限流窗口，瞬间打满每秒3次的额度
• 参考的异步任务重试方案：未接入限流逻辑，重试任务直接入队，违反限流规则；且pending.add(task.get_coro())存在语法错误（asyncio.wait要求pending是set[Task]而非协程对象）

核心实现思路 #

用全局双维度滑动窗口限流调度器统一拦截所有请求（包括重试的），确保只有满足「每秒≤3次、每小时≤1000次」的请求才会被执行；重试任务由队列重新调度，自动走限流校验，避免绕开规则。

第一步：实现双维度滑动窗口限流器 #

用deque存储最近的请求时间戳，配合异步条件变量处理并发等待，严格校验两个时间窗口的额度：

import asyncio
import time
from collections import deque
import logging

class DualRateLimiter:
    def __init__(self, short_limit: int, short_window: float, long_limit: int, long_window: float):
        # 短期限流：每秒3次
        self.short_limit = short_limit
        self.short_window = short_window
        # 长期限流：每小时1000次
        self.long_limit = long_limit
        self.long_window = long_window
        # 存储最近的请求时间戳（自动清理过期值）
        self.short_timestamps = deque()
        self.long_timestamps = deque()
        # 异步锁+条件变量，处理并发请求的等待/唤醒
        self.condition = asyncio.Condition()

    async def acquire(self):
        while True:
            async with self.condition:
                now = time.monotonic()  # 用单调时间避免系统时间修改的影响
                # 清理短期窗口的过期时间戳
                while self.short_timestamps and now - self.short_timestamps[0] > self.short_window:
                    self.short_timestamps.popleft()
                # 清理长期窗口的过期时间戳
                while self.long_timestamps and now - self.long_timestamps[0] > self.long_window:
                    self.long_timestamps.popleft()
                # 校验双窗口额度
                if len(self.short_timestamps) < self.short_limit and len(self.long_timestamps) < self.long_limit:
                    # 占用额度
                    self.short_timestamps.append(now)
                    self.long_timestamps.append(now)
                    return
                # 计算需要等待的最短时间（取最早过期的时间差）
                wait_time = min(
                    self.short_window - (now - self.short_timestamps[0]) if self.short_timestamps else 0,
                    self.long_window - (now - self.long_timestamps[0]) if self.long_timestamps else 0
                )
                # 异步等待额度释放
                await self.condition.wait_for(wait_time)

第二步：接入请求逻辑+可控重试 #

所有请求必须先经过限流校验，重试任务由队列重新调度（避免死循环，限制最大重试次数）：

# 初始化全局限流器：每秒3次、每小时1000次
rate_limiter = DualRateLimiter(short_limit=3, short_window=1, long_limit=1000, long_window=3600)
# 定义重试规则：最多重试3次，仅针对可重试异常（如API错误码29、网络中断）
MAX_RETRIES = 3
# 自定义可重试异常
class RetryableError(Exception):
    pass

async def limited_make_request(*args, **kwargs):
    # 所有请求必须先拿限流许可
    await rate_limiter.acquire()
    try:
        return await make_request(*args)
    except APIError as e:
        if e.code == 29:  # 仅针对指定错误码重试
            raise RetryableError(f"API error {e.code}") from e
        raise  # 非可重试异常直接抛出
    except Exception as e:
        # 处理网络中断等通用异常
        if "connection" in str(e).lower():
            raise RetryableError(f"Network error: {e}") from e
        raise

# 队列调度：处理任务的重试
async def run_tasks(coro_args_list: list) -> list:
    results = []
    # 待处理队列：(请求参数, 已重试次数)
    task_queue = asyncio.Queue()
    for args in coro_args_list:
        await task_queue.put((args, 0))

    # 并发worker：从队列取任务执行，限流器会自动拦截超额请求
    async def worker():
        while not task_queue.empty():
            try:
                args, retries = await task_queue.get_nowait()
            except asyncio.QueueEmpty:
                break
            try:
                res = await limited_make_request(*args)
                results.append(res)
            except RetryableError as e:
                if retries < MAX_RETRIES:
                    # 重试任务重新入队，自动走限流校验
                    await task_queue.put((args, retries + 1))
                    logging.warning(f"Retry request {args}, count: {retries+1}")
                else:
                    logging.error(f"Max retries exceeded for {args}: {e}")
            finally:
                task_queue.task_done()

    # 启动worker（数量可设为10以内，限流器会控制实际请求数）
    workers = [asyncio.create_task(worker()) for _ in range(10)]
    await asyncio.gather(*workers)
    return results

关键优化点 #

• 严格双维度限流：滑动窗口自动清理过期时间戳，所有请求（包括重试的）必须通过校验，完全符合规则
• 重试无绕权：重试任务由队列重新调度，不会直接发请求
• 异步高效等待：用asyncio.Condition替代固定sleep，额度释放时立即唤醒等待的任务
• 可控重试：限制最大重试次数，避免无限请求拖垮系统

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