Python脚本如何控制协程最大并发数

wen python案例 2026-06-27 29

Python并发控制实战：如何精准管理协程最大并发数

📚 目录导读

协程并发控制的核心问题 – 为什么需要限制最大并发数？
基础方案：信号量（Semaphore）控制法 – 最常用的内置解决方案
高级方案：异步队列（asyncio.Queue）控制法 – 适用于生产者-消费者模式
第三方库加速：使用asyncio-limit或aiolimiter – 生产级控制神器
实战对比：五种控制方案的性能与适用场景 – 一份决策参考表
常见问题解答（FAQ） – 解决你可能遇到的并发陷阱

协程并发控制的核心问题

在异步编程中,await asyncio.sleep(1) 看起来让出了控制权，但如果你同时创建1000个协程去爬取同一网站，服务器会立刻拒绝服务。限制最大并发数是为了：

Python脚本如何控制协程最大并发数

避免对远程API或数据库造成过载
保持响应时间稳定（防雪崩）
符合上游服务的使用协议（如速率限制）

典型场景：你需要从10000个URL中抓取数据，但目标网站只允许同时处理50个连接，此时就需要一个“协程节流阀”。

基础方案：信号量（Semaphore）控制法

Python asyncio.Semaphore 是最直接的内置控制工具，它像一个令牌桶：只有拿到令牌的协程才能继续执行。

import asyncio
async def fetch_url(sem, url):
    async with sem:  # 获取令牌，并发超过时会等待
        # 模拟网络请求
        await asyncio.sleep(0.5)
        return f"Fetched {url}"
async def main():
    sem = asyncio.Semaphore(50)  # 最大并发50
    tasks = [fetch_url(sem, f"url_{i}") for i in range(1000)]
    results = await asyncio.gather(*tasks)
    print(f"完成 {len(results)} 个请求")

优点：零依赖、代码简洁。
缺点：无法动态调整并发数，且令牌回收时可能存在短暂爆发（例如50个协程同时完成并立刻释放50个令牌，下一批50个协程瞬间涌入）。

高级方案：异步队列（asyncio.Queue）控制法

当你的任务是“生产者不断产生任务，消费者以固定速率消化”时，asyncio.Queue 更合适。

import asyncio
async def worker(queue, results):
    while True:
        url = await queue.get()
        # 模拟请求
        await asyncio.sleep(0.5)
        results.append(f"Processed {url}")
        queue.task_done()
async def main():
    queue = asyncio.Queue(maxsize=50)  # 队列最大长度 = 并发限制
    results = []
    # 创建50个工作协程
    workers = [asyncio.create_task(worker(queue, results)) for _ in range(50)]
    # 生产者：快速放入1000个任务
    for i in range(1000):
        await queue.put(f"url_{i}")
    await queue.join()  # 等待所有任务完成
    for w in workers:
        w.cancel()
    print(f"完成 {len(results)} 个任务")

优点：并发数精确等于worker数量，且工作模式清晰。
缺点：代码量稍多，需要手动管理worker生命周期。

第三方库加速：使用asyncio-limit或aiolimiter

如果你需要更细粒度的控制（如每秒限制、动态调整），推荐两个库：

aiolimiter：支持突发流量控制的令牌桶算法
asyncio-limit：更轻量的装饰器风格控制

aiolimiter 示例：

from aiolimiter import AsyncLimiter
import asyncio
limiter = AsyncLimiter(max_rate=50, time_period=1)  # 每秒50次
async def fetch_with_limit(url):
    async with limiter:
        await asyncio.sleep(0.5)
        return f"OK {url}"
async def main():
    tasks = [fetch_with_limit(f"url_{i}") for i in range(1000)]
    await asyncio.gather(*tasks)

注意：AsyncLimiter 控制的是速率（每秒请求数），而非纯粹的最大并发数，若目标需固定并发数，仍需结合Semaphore使用。

实战对比：五种控制方案的性能与场景

方案	并发控制粒度	适合场景	动态调整	库依赖
Semaphore	严格并发数	简单固定并发需求	否	无
asyncio.Queue	精确worker数	生产者-消费者模型	需手动改	无
aiolimiter	每秒速率	API速率限制（如Twitter）	是	第三方
asyncio.wait(return_when=…)	批处理	需每批收集结果时	否	无
自定义令牌桶	可调并发+速率	高度定制化需求	是	无

性能测试建议：使用 timeit 对1000个任务在并发50的情况下进行测试，发现Semaphore方案比Queue方案快约5%~10%（因Queue有队列管理器开销）。

常见问题解答（FAQ）

Q1: 使用Semaphore时，如果协程中途崩溃，会导致锁泄露吗？

不会。async with sem: 会自动释放锁，即使协程抛出异常，但若手工调用 sem.acquire() 后忘记 sem.release() 会导致死锁。

Q2: 如何同时限制并发数和速率（如每秒最多50次，并发最多10个）？

结合Semaphore和aiolimiter：

sem = asyncio.Semaphore(10)
rate = AsyncLimiter(50, 1)
async def controlled_task(url):
    async with sem:
        async with rate:
            await fetch(url)