DeepSeek网页版入口 · 在线免登录对话

无需本地算力显卡，打开浏览器即开即用。体验由 DeepSeek AI 驱动的自主反思推理引擎与代码重构力。

请帮我用 Python 编写一段高效的并行请求调度器，支持速率限制（Rate Limiter）与失败重试。

🧠 DeepSeek 思维链推理过程 (已用时 3.1 秒)

用户需求是构建面向高并发的异步速率限制器。采用令牌桶算法配合 asyncio 信号量机制，并在网络抖动时提供带抖动因子的指数退避重试……

针对您的高并发调度需求，推荐使用 asyncio + aiohttp 配合令牌桶速率限制器实现。核心代码结构如下：

import asyncio
import time

class AsyncRateLimiter:
    def __init__(self, rate: int, per: float = 1.0):
        self.rate = rate
        self.per = per
        self.semaphore = asyncio.Semaphore(rate)

    async def __aenter__(self):
        await self.semaphore.acquire()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        asyncio.create_task(self._release_after())

    async def _release_after(self):
        await asyncio.sleep(self.per)
        self.semaphore.release()

DECK STUB