Python 实战：基于 Asyncio 与令牌桶算法的高并发请求限流器实现

【技术专栏】 Python 核心实战系列
【分类标签】 并发编程与异步IO
【核心摘要】 深入探讨高并发爬虫与接口调用中的限流场景，结合 Asyncio 异步队列与令牌桶（Token Bucket）算法构建生产级限流器，有效防止请求被封禁与服务过载。

一、业务背景与实战痛点

在进行高并发 API 调用、微服务间通信或大规模数据采集时，下游服务往往有严格的 QPS（每秒请求数）限制。如果瞬间突发流量过大，极易触发 429 Too Many Requests 导致任务中断。传统的同步 sleep 会阻塞线程，无法充分发挥异步 IO 的性能优势。

二、核心设计思想与架构方案

本方案采用 **令牌桶算法（Token Bucket Algorithm）** 结合 Python 3 的 `asyncio` 原生并发原语。令牌桶以恒定速率向桶中填充令牌，请求进入时必须先获取令牌方可执行；若令牌不足则进入异步等待队列，既能允许短时间的突发流量，又能严格平滑全局并发速率。

三、完整生产级代码实现

以下为经过实战校验的完整核心代码，可直接集成进现有工程中：

import asyncio
import time

class AsyncTokenBucketLimiter:
    """基于 Asyncio 的高并发令牌桶限流器"""
    def __init__(self, rate: float, capacity: float):
        self.rate = float(rate)          # 令牌填充速率 (个/秒)
        self.capacity = float(capacity)  # 桶容量 (最大突发量)
        self.tokens = float(capacity)    # 当前令牌数量
        self.last_update = time.monotonic()
        self._lock = asyncio.Lock()

    async def _add_new_tokens(self):
        now = time.monotonic()
        elapsed = now - self.last_update
        self.last_update = now
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)

    async def acquire(self, tokens_needed: float = 1.0):
        """异步获取令牌，若不足则精确计算等待时间"""
        while True:
            async with self._lock:
                await self._add_new_tokens()
                if self.tokens >= tokens_needed:
                    self.tokens -= tokens_needed
                    return True
                deficit = tokens_needed - self.tokens
                wait_seconds = deficit / self.rate

            await asyncio.sleep(wait_seconds)

async def worker(task_id: int, limiter: AsyncTokenBucketLimiter):
    await limiter.acquire(1.0)
    print(f"[{time.strftime('%H:%M:%S')}] 任务 #{task_id:02d} 获得令牌并成功发起请求")

async def main():
    limiter = AsyncTokenBucketLimiter(rate=5.0, capacity=10.0)
    tasks = [worker(i, limiter) for i in range(20)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

四、避坑指南与生产实践建议

• **锁粒度控制**：计算等待时长后务必退出 `async with self._lock` 之后再执行 `asyncio.sleep`，否则会导致整个限流器死锁。
• **时间基准选择**：必须使用 `time.monotonic()` 单调递增时钟，严禁使用 `time.time()`，防止系统修改时间或夏令时变更造成令牌计算错乱。
• **分布式演进**：单机限流采用 asyncio.Lock，在多节点分布式集群中，可将令牌扣减逻辑迁移至 Redis Lua 脚本中原子执行。

五、总结与扩展思考

在企业级 Python 开发中，理解底层运行机制（如事件循环、内存管理、锁竞争与网络 I/O）是编写高性能、高可用代码的关键。建议读者在本地环境实际运行上述代码并进行压力测试与边界模拟，加深对架构设计的理解。

发布时间：2026年09月04日 | 华为云开发者社区技术分享