time.sleep()限流仅适用于单线程脚本，多线程下各线程独立休眠导致QPS失控；ratelimit库默认线程本地计数，需配合锁和全局计数器；异步环境应使用asyncio.Semaphore与后台补令牌任务；生产环境多实例必须用Redis+Lua原子操作，避免超卖。
用 time.sleep() 做简单限流，为什么常出问题
它确实能卡住请求节奏，但只适合单进程、无并发的脚本。一旦加了多线程或多进程，time.sleep() 只阻塞当前线程，其他线程照发不误，实际 QPS 完全失控。
常见错误现象：requests.get() 在循环里配了 time.sleep(1)，结果压测发现每秒发出 5–6 个请求——因为开了 5 个线程，每个都睡自己的 1 秒，互不感知。
- 适用场景：调试用的单次脚本、CLI 工具的简单重试延时
- 千万别用在
ThreadPoolExecutor或asyncio环境里 - 没有状态共享，无法统计“过去 10 秒发了多少个”，也就谈不上滑动窗口
ratelimit 库的 @sleep_and_retry + @limits 组合怎么配才不翻车
这个组合看似开箱即用，但默认是基于线程本地计数器的，多线程下会各自维护一套计数，导致总请求数超标。
必须显式传入 key_func 和共享存储（比如用 threading.Lock + 全局 dict），否则在 concurrent.futures 场景下大概率超限。
立即学习“Python免费学习笔记（深入）”；
- 正确写法示例：
from ratelimit import limits, sleep_and_retry import threading _counter = {"count": 0, "last_reset": 0} _lock = threading.Lock() def key_func(*args, **kwargs): return "global" @sleep_and_retry @limits(calls=10, period=60, key_func=key_func) def call_api(): with _lock: _counter["count"] += 1 - 注意
period单位是秒，不是毫秒；calls是整数，不支持小数（比如 0.2 次/秒得换算法） - 该库不兼容
asyncio，异步客户端必须换方案
异步 HTTP 客户端（aiohttp）怎么实现令牌桶
异步环境不能靠锁和全局变量硬扛，得用 asyncio.Semaphore 配合时间戳做令牌生成逻辑。核心是每次请求前检查“当前令牌数 ≥ 1”，不够就按缺额等待。
容易踩的坑：直接用 asyncio.sleep() 等待会导致整个 event loop 延迟，应改用 await asyncio.wait_for(sem.acquire(), timeout=...) 配合定期 refill 任务。
- 推荐结构：启动一个后台
asyncio.create_task(refill_loop())，每 100ms 补 1 个令牌（根据速率换算） -
sem = asyncio.Semaphore(0)初始化为 0，首次 acquire 必然挂起，等 refill 启动后才放行 - 别把 refill 逻辑写进请求函数里——高频调用会导致时间戳判断失准和重复补发
生产环境该选 redis-py 还是内存计数器
单机服务且不惧重启丢数据，用 threading.local 或 functools.lru_cache 配时间窗口最快；但只要涉及多实例、滚动发布或需要精确削峰，就必须上 Redis。
Redis 方案不是简单 INCR + EXPIRE 就完事——Lua 脚本才是原子关键。漏掉这步，高并发下 GET + INCR + SETEX 三步分离会引发超卖。
- 必须用 Lua 做原子操作，例如：
local current = redis.call("INCR", KEYS[1]) if current == 1 then redis.call("EXPIRE", KEYS[1], ARGV[1]) end return current - 客户端要处理
ConnectionError和TimeoutError：降级为内存限流 or 直接放过，别让限流组件拖垮主链路 - Redis 的
TIMEOUT设置建议比窗口周期长 10%，避免刚好过期时新请求被误判
真正难的从来不是“怎么写个限流器”，而是“怎么让它的行为在故障、扩容、时钟跳变时不背叛你的 SLA”。这些边界条件，往往在压测之后才浮出来。
