time.sleep()在异步编程中会冻结事件循环，导致所有协程、io和定时器停摆；必须改用await asyncio.sleep()让出控制权，或通过run_in_executor处理同步阻塞调用。
用 time.sleep() 硬等是最容易出错的方案
很多人一想到“限频”就直接在循环里加 time.sleep(1)，结果发现：请求没卡住，反而越跑越快，甚至触发目标站的突发流量检测。根本原因是 time.sleep() 只阻塞当前线程，对异步协程（asyncio）完全无效，而且无法应对多任务并发场景下的全局速率控制。
真正需要的是带状态的、可共享的节流机制。常见错误包括：
- 每个请求单独计时，没统一窗口（比如“每秒最多 5 次”，不是“每次间隔 ≥200ms”）
- 没考虑网络超时或重试导致的重复计数
- 多进程下用普通变量计数，状态不共享
推荐用 ratelimit 库做同步限流
轻量、无依赖、API 直观，适合 requests + 线程池场景。它底层用滑动窗口+时间戳判断，比简单计数更准。
安装与基本用法：
pip install ratelimit
关键点：
-
@rate_limited(calls=5, period=1)装饰器作用于函数，自动按窗口限频 - 默认使用
threading.Lock，多线程安全；但多进程需配合multiprocessing.Manager或换用 Redis 后端 - 被限流时会抛出
ratelimit.exception.RateLimitException，建议显式捕获而非忽略
示例：
from ratelimit import limits, sleep_and_retry import requests <p>@sleep_and_retry @limits(calls=5, period=1) def fetch_url(url): return requests.get(url, timeout=5) </p>
异步请求必须用 aiolimiter 或自建 asyncio.Semaphore + 时间窗
asyncio.Semaphore(5) 只能控并发数（同时最多 5 个请求），不能控频率（比如“5 QPS”）。要实现真正的异步限频，得结合时间窗口判断。
aiolimiter 是目前最稳妥的选择：
- 支持滑动窗口和固定窗口两种模式
- 内部用
asyncio.Lock和asyncio.wait_for实现等待，不阻塞事件循环 - 注意：它的
acquire()是协程，必须await，不能漏掉
示例：
from aiolimiter import AsyncLimiter import aiohttp import asyncio <p>limiter = AsyncLimiter(5, 1) # 5 calls / second</p><p>async def fetch(session, url): await limiter.acquire() async with session.get(url) as resp: return await resp.text() </p>
生产环境建议加一层 Redis 计数器兜底
单机限频在分布式爬虫或容器化部署中会失效——多个实例各自计数，总请求数就超标了。这时必须引入外部协调服务。
Redis 是最常用选择，用 INCR + EXPIRE 组合实现原子性窗口计数：
- key 设为
f"{prefix}:{int(time.time() // window)}"（固定窗口），或用 Lua 脚本实现滑动窗口 - 务必设置过期时间，避免 key 持久堆积
- 网络抖动时 Redis 不可用，需有降级策略（如 fallback 到本地内存限频 + 日志告警）
不要手写 Redis 限频逻辑，优先用 redis-cell 模块（基于 Redis 的漏斗算法模块）或封装好的 slowapi（虽为 FastAPI 设计，但核心限流逻辑可复用）。
真正难的不是写限频代码，而是确认目标站点的真实容忍阈值——有些站对 User-Agent 频率敏感，有些对 IP+Cookie 组合敏感，还有些会动态调整阈值。留足余量、加随机 jitter、监控返回状态码分布，比选哪个库更重要。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
