Python脚本如何限制异步任务并发数

本文目录导读：

1. 使用 asyncio.Semaphore（最常用）
2. 使用 asyncio.Queue + worker模式
3. 使用 asyncio.wait + 分批处理
4. 使用 aiometer 库（第三方）
5. 通用装饰器方式
6. 选择建议

在Python中限制异步任务并发数,最常用的方法是使用asyncio.Semaphore（信号量），以下是几种常见的实现方式：

使用 asyncio.Semaphore（最常用）

import asyncio
import aiohttp
async def fetch_url(semaphore, url):
    async with semaphore:  # 获取信号量，限制并发
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.text()
async def main():
    urls = ["http://example.com"] * 20
    semaphore = asyncio.Semaphore(5)  # 限制并发数为5
    # 创建所有任务
    tasks = [fetch_url(semaphore, url) for url in urls]
    # 并发执行，但受信号量限制
    results = await asyncio.gather(*tasks)
    return results
# 运行
asyncio.run(main())

使用 asyncio.Queue + worker模式

import asyncio
async def worker(queue, semaphore):
    while True:
        url = await queue.get()
        if url is None:  # 结束信号
            break
        async with semaphore:
            # 执行实际的任务
            print(f"Processing: {url}")
            await asyncio.sleep(1)  # 模拟异步操作
        queue.task_done()
async def main():
    urls = [f"url_{i}" for i in range(20)]
    queue = asyncio.Queue()
    semaphore = asyncio.Semaphore(5)
    # 创建worker（并发数为5）
    workers = [asyncio.create_task(worker(queue, semaphore)) 
               for _ in range(5)]
    # 添加任务
    for url in urls:
        await queue.put(url)
    # 等待所有任务完成
    await queue.join()
    # 发送停止信号
    for _ in workers:
        await queue.put(None)
    # 等待worker结束
    await asyncio.gather(*workers)
asyncio.run(main())

使用 asyncio.wait + 分批处理

import asyncio
async def process_item(item):
    await asyncio.sleep(1)
    return f"Processed: {item}"
async def limited_concurrent(items, limit=5):
    results = []
    # 分批处理
    for i in range(0, len(items), limit):
        batch = items[i:i+limit]
        tasks = [process_item(item) for item in batch]
        batch_results = await asyncio.gather(*tasks)
        results.extend(batch_results)
    return results
# 使用
items = list(range(20))
results = asyncio.run(limited_concurrent(items, 5))

使用 aiometer 库（第三方）

首先安装：pip install aiometer

import asyncio
import aiometer
async def process_item(item):
    await asyncio.sleep(1)
    return f"Processed: {item}"
async def main():
    items = list(range(20))
    # 限制并发数为5
    async with aiometer.amap(
        process_item,
        items,
        max_at_once=5,  # 同时执行的最大数量
        max_per_second=10  # 每秒执行的最大数量（可选）
    ) as results:
        async for result in results:
            print(result)
asyncio.run(main())

通用装饰器方式

import asyncio
from functools import wraps
def limit_concurrency(limit):
    semaphore = asyncio.Semaphore(limit)
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            async with semaphore:
                return await func(*args, **kwargs)
        return wrapper
    return decorator
# 使用装饰器
@limit_concurrency(5)
async def fetch_data(url):
    await asyncio.sleep(1)
    return f"Data from {url}"
async def main():
    urls = [f"url_{i}" for i in range(20)]
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    return results
asyncio.run(main())

选择建议

• 简单场景：使用 asyncio.Semaphore，最直接
• 需要控制输入输出流：使用 asyncio.Queue
• 需要限速：使用 aiometer 库
• 代码复用：使用装饰器方式

每种方法都有其适用场景,根据具体需求选择合适的实现方式。