asyncio.semaphore仅限制并发数，不控制时间窗口内的请求频率，无法实现“每秒n次”的限流需求；真正可控的频率限制需用滑动窗口或令牌桶算法，如自定义ratelimiter类配合asyncio.queue与time.monotonic()。
为什么 asyncio.Semaphore 不能直接限制请求频率
很多人以为 asyncio.Semaphore 能像限流器（rate limiter）一样控制“每秒最多 N 次调用”，其实它只管并发数，不管时间窗口。比如设 semaphore = asyncio.Semaphore(3)，它允许最多 3 个协程同时进入临界区，但不阻止这 3 个协程在 10ms 内全发出去——这和“每秒最多 5 次 HTTP 请求”的业务需求根本不是一回事。
常见错误现象：asyncio.Semaphore 配合 asyncio.gather 大量任务时，QPS 爆表、被目标服务 429 拒绝，但日志里看不出并发超了（因为确实没超 semaphore 的计数）。
- 它解决的是资源竞争（如共享连接池、文件句柄），不是节流（throttling）
- 它不感知时间，无法实现“滑动窗口”或“令牌桶”语义
- 若混用
sleep手动凑间隔，容易因协程调度偏差导致实际 QPS 不稳
真正可控的并发频率：用 asyncio.Queue + 时间戳滑动窗口
要实现“每秒最多 N 次”，得自己维护一个带时间戳的请求队列，并在每次准入前检查窗口内已发生的请求数。比 Semaphore 更可靠，也比第三方库（如 aiolimiter）更透明、无隐藏行为。
核心思路：用 asyncio.Queue(maxsize=N) 做令牌桶的“桶”，配合 time.time() 判断上一次放行是否在窗口内；若桶满且未过期，则 await asyncio.sleep() 等待空位。
import asyncio
import time
<p>class RateLimiter:
def <strong>init</strong>(self, max_calls: int, period: float = 1.0):
self.max_calls = max_calls
self.period = period
self.calls = [] # [(timestamp, ...)]</p><pre class="brush:python;toolbar:false;">async def acquire(self):
now = time.time()
# 清理过期记录
self.calls = [t for t in self.calls if now - t = self.max_calls:
# 算出最早一条多久后过期，就睡多久
sleep_time = self.calls[0] + self.period - now
if sleep_time > 0:
await asyncio.sleep(sleep_time)
# 再清理一遍（防止 sleep 期间又有新请求进来）
now = time.time()
self.calls = [t for t in self.calls if now - t
- 注意：这个类本身不是线程安全的，但在单个 event loop 内协程间是安全的
-
period设为 0.1 表示“每 100ms 最多 N 次”，适合高频低延迟场景 - 避免用
deque(maxlen=N)替代列表——它会丢弃旧时间戳，导致窗口计算错误
与 aiohttp 配合时，别在 session 层加 Semaphore
有人习惯给 aiohttp.ClientSession 加 connector=TCPConnector(limit=10)，再套一层 asyncio.Semaphore(10)，以为双重保险。结果反而拖慢整体吞吐：connector 限的是 TCP 连接数，Semaphore 限的是协程并发，两者作用域不同，叠加使用会导致“连接空闲但协程卡在 semaphore 等待”。
- 真实瓶颈常在 DNS 解析或 TLS 握手，
limit应略高于目标 QPS（比如 QPS=50，设limit=60） - 如果目标 API 明确要求“每秒 20 次”，应把
RateLimiter(20, 1.0)放在 request 发起前，而不是 session 初始化时 - 不要对整个
session.get()包裹semaphore.acquire()，而应在构造 URL 后、调用前 acquire，在 response 处理完后 release（或用 async context manager）
生产环境必须处理的三个边界情况
限流逻辑一旦上线，就会暴露平时测试不到的问题：
- 系统时间回拨：如果服务器 NTP 同步导致
time.time()突然变小，滑动窗口可能误判“大量请求堆积”，应改用time.monotonic()做窗口判断 - 协程取消：用户
cancel()一个正在await limiter.acquire()的任务，需确保acquire支持 cancel 并清理状态（当前示例不支持，需加try/finally和self.calls.remove()回滚） - 跨进程限流失效：单机
RateLimiter无法约束多 worker（如 uvicorn 多进程），此时必须上 Redis + Lua 原子操作，或改用分布式限流中间件
最常被忽略的一点：限流不是越严越好。压测时发现下游响应延迟升高，应优先降 QPS 而非加重重试逻辑——后者会让雪崩来得更快。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
