3. Python 限流核心库介绍与基本使用

在金融 API 限流熔断框架中，选择合适的限流实现方式是架构设计的关键决策点。Python 生态提供了从基础同步原语到专用限流库的多种选择，本节将系统介绍 time.sleep、threading.Timer、ratelimit 库和 pyrate-limiter 库的安装与基本使用，并对比其适用场景。

3.1 四种限流方式概览

3.2 time.sleep：最基础的阻塞限流

time.sleep(seconds) 是 Python 内置的线程阻塞函数，通过让当前线程暂停执行指定秒数来实现限流。其核心逻辑是：在每次 API 调用后插入固定延迟，从而控制单位时间内的请求频率。

import time
import requests

def call_api_with_sleep(url, min_interval=1.0):
    """使用 time.sleep 实现每秒最多 1 次请求"""
    response = requests.get(url)
    time.sleep(min_interval)  # 阻塞当前线程
    return response

# 模拟连续调用
for i in range(5):
    result = call_api_with_sleep("https://api.example.com/quote")
    print(f"请求 {i+1} 完成，状态码: {result.status_code}")

3.3 threading.Timer：非阻塞延迟触发

threading.Timer 创建一个新线程，在指定延迟后执行回调函数。它不阻塞主线程，适合需要定时触发限流重置或定时任务的场景。

import threading
import time

def reset_rate_limit():
    """模拟限流计数器重置"""
    print(f"[{time.strftime('%H:%M:%S')}] 限流计数器已重置")

# 每 5 秒执行一次重置
timer = threading.Timer(5.0, reset_rate_limit)
timer.start()

print(f"[{time.strftime('%H:%M:%S')}] 主线程继续执行其他任务...")
# 主线程不阻塞

在限流框架中，threading.Timer 常用于实现令牌桶的定时补充逻辑，或滑动窗口的过期清理任务。

3.4 ratelimit 库：装饰器风格的函数限流

ratelimit 库通过装饰器为函数添加调用频率限制，支持 calls（最大调用次数）和 period（时间窗口）两个核心参数。安装命令：

pip install ratelimit

基本使用示例：

from ratelimit import limits, sleep_and_retry
import requests

# 配置：每分钟最多 10 次调用
@sleep_and_retry
@limits(calls=10, period=60)
def fetch_market_data(symbol):
    """获取市场行情数据（受限流保护）"""
    response = requests.get(f"https://api.example.com/market/{symbol}")
    return response.json()

# 测试调用
for symbol in ["AAPL", "GOOGL", "MSFT", "AMZN", "TSLA"]:
    data = fetch_market_data(symbol)
    print(f"{symbol}: {data['price']}")

3.5 pyrate-limiter 库：多桶策略与灵活配置

pyrate-limiter 是一个功能更全面的限流库，支持令牌桶、漏桶和滑动窗口等多种算法，并允许配置多个限流维度。安装命令：

pip install pyrate-limiter

基本使用示例（令牌桶模式）：

from pyrate_limiter import Duration, RequestRate, Limiter
import time

# 定义限流规则：每秒 5 个令牌，每分钟 100 个令牌
rate_per_second = RequestRate(5, Duration.SECOND)
rate_per_minute = RequestRate(100, Duration.MINUTE)

# 创建限流器
limiter = Limiter(rate_per_second, rate_per_minute)

# 模拟 API 调用
for i in range(20):
    try:
        limiter.try_acquire("api_call")
        print(f"请求 {i+1}: 通过")
    except Exception as e:
        print(f"请求 {i+1}: 被限流 - {e}")
    time.sleep(0.1)

3.6 四种限流方式对比 SVG 架构图

3.7 选型决策流程

在实际金融项目中，选择限流库需要综合考虑以下因素：

1. 精度要求： 高精度场景（如每秒精确控制）选择 pyrate-limiter；一般场景 ratelimit 足够
2. 并发模型： 异步框架（asyncio）避免使用 time.sleep，优先选择支持异步的限流库
3. 多维度限流： 需要按用户、接口、IP 分别限流时，pyrate-limiter 的多桶策略是首选
4. 运维复杂度： 简单场景优先使用标准库减少依赖，复杂场景引入专用库