在Python异步编程场景中，无限制地创建异步任务很容易引发资源占用过高、接口请求被封禁等问题，因此需要合理控制异步任务的并发数量。常见的控制方案可以基于asyncio原生功能实现，也可以借助第三方库完成，下面逐一介绍具体的实现方式。
方案一：使用asyncio.Semaphore信号量控制
asyncio内置的Semaphore是最常用的并发控制组件，它通过维护一个计数器来限制同时运行的任务数量，计数器初始值为允许的最大并发数，任务执行前需要获取信号量，执行完成后释放信号量。
基础实现示例
import asyncio
# 定义最大并发数为3
MAX_CONCURRENCY = 3
semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
async def async_task(task_id):
# 获取信号量，若已达最大并发则阻塞等待
async with semaphore:
print(f"任务{task_id}开始执行")
# 模拟任务执行耗时
await asyncio.sleep(2)
print(f"任务{task_id}执行完成")
return f"任务{task_id}结果"
async def main():
# 创建10个待执行的异步任务
tasks = [async_task(i) for i in range(10)]
# 等待所有任务完成
results = await asyncio.gather(*tasks)
print("所有任务执行结果:", results)
if __name__ == "__main__":
asyncio.run(main())
上述代码中，信号量初始值设为3，因此同一时间最多只有3个任务可以同时执行，后续的7个任务会等待前面的任务释放信号量后再依次执行，有效避免了并发数过高的问题。
方案二：使用asyncio.Queue实现任务队列控制
除了信号量，还可以通过asyncio.Queue构建任务队列，预先将任务放入队列，再启动固定数量的消费者协程从队列中取任务执行，消费者协程的数量就是最大并发数。
实现示例
import asyncio
async def task_worker(queue, worker_id):
while True:
# 从队列中获取任务，队列为空则阻塞等待
task_id = await queue.get()
print(f"工作者{worker_id}开始处理任务{task_id}")
await asyncio.sleep(2)
print(f"工作者{worker_id}处理任务{task_id}完成")
# 标记任务已完成
queue.task_done()
async def main():
# 定义最大并发数为4，即启动4个工作者协程
WORKER_NUM = 4
task_queue = asyncio.Queue()
# 启动工作者协程
workers = [asyncio.create_task(task_worker(task_queue, i)) for i in range(WORKER_NUM)]
# 向队列中添加10个任务
for task_id in range(10):
await task_queue.put(task_id)
# 等待队列中所有任务处理完成
await task_queue.join()
# 取消工作者协程，避免程序无法退出
for worker in workers:
worker.cancel()
print("所有任务执行完成")
if __name__ == "__main__":
asyncio.run(main())
这种方案的优势是可以灵活控制任务的生产和消费节奏，适合任务数量动态变化的场景，比如需要实时接收外部任务并控制并发处理的场景。
方案三：使用第三方库aiometer控制并发
如果不想自己实现并发控制逻辑，可以使用第三方库aiometer，它提供了更简洁的API来控制异步任务的并发数、速率等参数，适合快速开发场景。
安装与使用示例
首先通过pip安装aiometer：
pip install aiometer
使用示例代码如下：
import asyncio
import aiometer
async def async_task(task_id):
print(f"任务{task_id}开始执行")
await asyncio.sleep(2)
print(f"任务{task_id}执行完成")
return f"任务{task_id}结果"
async def main():
# 控制最大并发数为3，执行10个任务
results = await aiometer.run_on_each(
async_task,
range(10),
max_at_once=3
)
print("所有任务执行结果:", results)
if __name__ == "__main__":
asyncio.run(main())
aiometer还支持设置最大每秒执行的任务数等参数，功能比原生信号量更丰富，适合需要更精细控制任务执行节奏的场景。
不同方案的选择建议
如果项目中没有额外的第三方库依赖要求，优先选择asyncio.Semaphore方案，原生支持不需要额外安装依赖，逻辑也足够清晰。如果任务的生产和消费是分离的，需要更灵活的任务调度，可以选择asyncio.Queue方案。如果希望快速实现且需要更多控制参数，可以选择aiometer第三方库。
| 方案 | 优势 | 适用场景 |
|---|---|---|
| asyncio.Semaphore | 原生支持，无额外依赖，逻辑简单 | 简单的并发数控制场景 |
| asyncio.Queue | 支持任务动态调度，生产消费解耦 | 任务数量动态变化、需要灵活调度的场景 |
| aiometer | API简洁，支持速率控制等更多参数 | 快速开发、需要精细控制执行节奏的场景 |
注意：控制并发数时需要结合实际的系统资源、目标服务的限流规则来设置合理的最大并发值，避免设置过高或过低影响程序性能。