Python异步编程中asyncio.gather的并发控制的介绍_F11

APP正在开发中...

返回顶部

分享到

Python异步编程中asyncio.gather的并发控制的介绍

python 来源：互联网作者：佚名发布时间：2025-03-24 22:15:18 人浏览

摘要

在Python异步编程生态中，asyncio.gather是并发任务调度的核心工具。然而当面对海量任务时，不加控制的并发可能引发资源耗尽、服务降级等问题。本文将通过实际场景和代码示例，展示如何结合

在Python异步编程生态中，asyncio.gather是并发任务调度的核心工具。然而当面对海量任务时，不加控制的并发可能引发资源耗尽、服务降级等问题。本文将通过实际场景和代码示例，展示如何结合信号量机制实现精准并发控制，既保证吞吐量又避免系统过载。

一、asyncio.gather的原始行为解析

asyncio.gather的设计初衷是批量执行异步任务，其默认行为类似于"全速冲刺"：

import asyncio

async def task(n):

print(f"Task {n} started")

await asyncio.sleep(1)

print(f"Task {n} completed")

return n

async def main():

tasks = [task(i) for i in range(10)]

results = await asyncio.gather(*tasks)

print(f"Total results: {len(results)}")

asyncio.run(main())

在这个示例中，10个任务会立即全部启动，1秒后几乎同时完成。这种"全并发"模式在以下场景存在隐患：

网络请求：同时发起数千个HTTP请求可能被目标服务器封禁

文件IO：磁盘IO密集型操作会拖慢系统响应

数据库连接：超过连接池限制导致报错

二、信号量控制法：给并发装上"节流阀"

asyncio.Semaphore通过限制同时执行的任务数，实现精准并发控制。其核心机制是：

初始化时设定最大并发数（如10）

每个任务执行前必须获取信号量

任务完成后释放信号量

async def controlled_task(sem, n):

async with sem: # 获取信号量

print(f"Task {n} acquired semaphore")

await asyncio.sleep(1)

print(f"Task {n} released semaphore")

return n

async def main():

sem = asyncio.Semaphore(3) # 最大并发3

tasks = [controlled_task(sem, i) for i in range(10)]

results = await asyncio.gather(*tasks)

print(f"Total results: {len(results)}")

asyncio.run(main())

执行效果：

始终只有3个任务在执行
每完成1个任务，立即启动新任务
总耗时≈4秒（10/3向上取整）

三、进阶控制策略

3.1 动态调整并发数

通过监控队列长度动态调整信号量：

async def dynamic_control():

sem = asyncio.Semaphore(5)

task_queue = asyncio.Queue()

# 生产者

async def producer():

for i in range(20):

await task_queue.put(i)

# 消费者

async def consumer():

while True:

item = await task_queue.get()

async with sem:

print(f"Processing {item}")

await asyncio.sleep(1)

task_queue.task_done()

# 动态调整

def monitor(queue):

while True:

size = queue.qsize()

if size > 10:

sem._value = max(1, sem._value - 1)

elif size < 5:

sem._value = min(10, sem._value + 1)

asyncio.sleep(1)

await asyncio.gather(

producer(),

*[consumer() for _ in range(3)],

asyncio.to_thread(monitor, task_queue)

)

asyncio.run(dynamic_control())

3.2 分批执行策略

对于超大规模任务集，可采用分批处理：

def chunked(iterable, chunk_size):

for i in range(0, len(iterable), chunk_size):

yield iterable[i:i+chunk_size]

async def batch_processing():

all_tasks = [task(i) for i in range(100)]

for batch in chunked(all_tasks, 10):

print(f"Processing batch: {len(batch)} tasks")

await asyncio.gather(*batch)

asyncio.run(batch_processing())

优势：

避免内存爆炸
方便进度跟踪
支持中间状态保存

四、性能对比与最佳实践

控制方式	吞吐量	资源占用	实现复杂度	适用场景
无控制	高	高	低	小型任务集
固定信号量	中	中	中	通用场景
动态信号量	中高	中低	高	需要弹性控制的场景
分批处理	低	低	中	超大规模任务集

最佳实践建议：

网络请求类任务：并发数控制在5-20之间

文件IO操作：并发数不超过CPU逻辑核心数*2

数据库操作：遵循连接池最大连接数限制

始终设置合理的超时时间：

try:

await asyncio.wait_for(task(), timeout=10)

except asyncio.TimeoutError:

print("Task timed out")

五、常见错误与解决方案

错误1：信号量未正确释放

# 错误示例：缺少async with

sem = asyncio.Semaphore(3)

sem.acquire()

await task()

sem.release() # 容易忘记释放

解决方案：

# 正确用法

async with sem:

await task() # 自动获取和释放

错误2：任务异常导致信号量泄漏

async def risky_task():

async with sem:

raise Exception("Oops!") # 异常导致sem未释放

解决方案：

async def safe_task():

sem_acquired = False

try:

async with sem:

sem_acquired = True

# 执行可能出错的操作

finally:

if not sem_acquired:

sem.release()

结语

asyncio.gather配合信号量机制，就像给异步程序装上了智能节流阀。通过合理设置并发参数，既能让程序高效运行，又能避免系统过载。实际开发中应根据任务类型、资源限制和SLA要求，选择最合适的并发控制策略。记住：优秀的并发控制不是追求最大速度，而是找到性能与稳定性的最佳平衡点。

您可能感兴趣的文章 :

原文链接 :

Tag : python(1096)asyncio(3)

Python中的异步与同步深度解析

Python中的异步与同步：深度解析与实践在Python编程世界里，异步和同步的概念是理解程序执行流程和性能优化的关键。这篇文章将带你深入
Python异步编程中asyncio.gather的并发控制的介绍

在Python异步编程生态中，asyncio.gather是并发任务调度的核心工具。然而当面对海量任务时，不加控制的并发可能引发资源耗尽、服务降级等问
基于Python打造一个高效开发辅助全能工具箱

在日常开发过程中，我们经常需要进行各种琐碎但又必不可少的操作，比如文件处理、编码转换、哈希计算、二维码生成、单位换算等。如
python中time模块的常用方法及应用

一、时间基石：time.time() time.time()是获取时间戳的入口函数，返回自1970年1月1日（Unix纪元）以来的秒数（浮点数）。这个10位数字像时间维
讯飞webapi语音识别接口调用代码(python)

基于python3 讯飞webAPI语音识别接口调用的实现。一、环境 1、注册讯飞平台账号：讯飞平台网址 2、进入控制台并创建应用 3、进入应用后点
Python不支持中文路径的解决方法

在编程的世界里，遇到问题并不罕见，但有些问题可能会让人感到格外棘手。比如，你是否曾经在使用Python处理文件时，因为路径中包含中
Python使用DeepSeek进行联网搜索功能

在当今信息爆炸的时代，联网搜索已成为获取数据、优化模型效果的重要手段。Python作为一种非常流行的编程语言，结合DeepSeek这一高性能的
Python操作Word文档7种方法的实现与对比(史上最全

在Python中操作Word文档是一项常见的任务，特别是在办公自动化和数据处理领域。本文将详细总结和对比几种常用的Python库和方法，包括它们
Python实现语音启动电脑应用程序

实现思路 osk模型进行输入语音转换 txt字典导航程序路径 pyttsx3引擎进行语音打印输出关键词=程序路径完整代码 1 2 3 4 5 6 7 8 9 10 11 12 13 14
Python实现高精度敏感词过滤

一、需求分析：敏感词过滤的四大核心挑战准确性要求：误判率需低于0.01% 性能压力：单机QPS需达到10万+ 对抗升级：需识别变体、拼音、谐

Python异步编程中asyncio.gather的并发控制的介绍

一、asyncio.gather的原始行为解析

二、信号量控制法：给并发装上"节流阀"

三、进阶控制策略

3.1 动态调整并发数

3.2 分批执行策略

四、性能对比与最佳实践

五、常见错误与解决方案

结语

您可能感兴趣的文章 :

Python中的异步与同步深度解析

Python异步编程中asyncio.gather的并发控制的介绍

基于Python打造一个高效开发辅助全能工具箱

python中time模块的常用方法及应用

讯飞webapi语音识别接口调用代码(python)

Python不支持中文路径的解决方法

Python使用DeepSeek进行联网搜索功能

Python操作Word文档7种方法的实现与对比(史上最全

Python实现语音启动电脑应用程序

Python实现高精度敏感词过滤

python批量下载抖音视频

利用Pyecharts可视化微信好友的方法

python爬取豆瓣电影TOP250数据

基于tensorflow权重文件的解读

解决Python字典查找报Keyerror的问题