
本文深入探讨了在Python asyncio中调度嵌套异步函数时遇到的并发挑战。通过分析传统`await`操作的阻塞特性,揭示了其在复杂流处理场景中的局限性。文章提出并详细阐述了基于`asyncio.Queue`和`asyncio.Event`的生产者-消费者模式,作为实现任务间解耦和真正并发执行的有效策略,从而显著提升异步应用的响应性和效率。
在异步编程中,我们经常需要处理数据流,其中一个任务负责生成数据,另一个任务负责处理数据。Python的asyncio库提供了强大的工具来构建并发应用程序,但在调度嵌套异步函数时,如果不正确地理解await关键字的行为,可能会导致程序并非按预期并发执行,而是串行阻塞。
考虑一个常见的场景:我们有一个字符流生成器,它逐步产生字符;一个句子生成器,它从字符流中收集字符并形成完整的句子;以及一个句子处理器,它对每个句子执行耗时操作。
以下是初始实现的代码结构:
import asyncio
async def stream():
char_string = "Hi. Hello. Hello."
for char in char_string:
await asyncio.sleep(0.1) # 模拟耗时操作
print("got char:", char)
yield char
async def sentences_generator():
sentence = ""
async for char in stream():
sentence += char
if char in [".", "!", "?"]:
print("got sentence: ", sentence)
yield sentence
sentence = ""
async def process_sentence(sentence: str):
print("waiting for processing sentence: ", sentence)
await asyncio.sleep(len(sentence)*0.1) # 模拟耗时操作
print("sentence processed!")
async def main():
i = 0
async for sentence in sentences_generator():
print("processing sentence: ", i)
await process_sentence(sentence) # 这里的await是关键
i += 1
asyncio.run(main())运行上述代码,其输出大致如下:
got char: H got char: i got char: . got sentence: Hi. processing sentence: 0 waiting for processing sentence: Hi. sentence processed! got char: got char: H got char: e got char: l got char: l got char: o got char: . got sentence: Hello. processing sentence: 1 waiting for processing sentence: Hello. sentence processed!
从输出可以看出,当process_sentence函数被await时,main协程会暂停,直到process_sentence完全执行完毕。这意味着在process_sentence处理第一个句子期间,stream和sentences_generator无法继续生成新的字符和句子。这并非我们期望的并发行为,我们希望在处理一个句子的同时,上游的字符流能够继续生成,从而提高整体吞吐量。
造成这种现象的根本原因在于await关键字的语义。当一个协程await另一个协程时,它会暂停自身的执行,并将控制权交给被await的协程。只有当被await的协程完成或自身也await了其他操作时,控制权才可能回到调用者。因此,在上述例子中,main函数中的await process_sentence(sentence)会完全阻塞main函数,直到当前句子处理完毕,才能继续从sentences_generator中获取下一个句子。
为了实现真正的并发,即在process_sentence处理句子的同时,sentences_generator和stream能够继续生成数据,我们可以采用经典的生产者-消费者模式。
在这种模式中:
asyncio提供了asyncio.Queue来实现这种异步安全的共享队列。此外,为了优雅地处理生产者完成后的消费者关闭问题,我们还可以引入asyncio.Event来发出生产者完成的信号。
Sitekick
一个AI登陆页面自动构建器
121
查看详情
以下是使用生产者-消费者模式重构后的代码:
import asyncio
async def stream():
char_string = "Hi. Hello. Thank you." # 更改了字符串以展示更长的流
for char in char_string:
await asyncio.sleep(0.1)
print("got char:", char)
yield char
async def sentences_generator(q: asyncio.Queue[str], flag: asyncio.Event):
"""
生产者:从字符流生成句子,并放入队列。
当字符流结束时,设置Event标志通知消费者。
"""
sentence = ""
async for char in stream():
sentence += char
if char in [".", "!", "?"]:
print("got sentence: ", sentence)
await q.put(sentence) # 将生成的句子放入队列
sentence = ""
flag.set() # 生产者完成所有句子的生成,设置Event
async def process_sentence(q: asyncio.Queue[str], flag: asyncio.Event):
"""
消费者:从队列中取出句子进行处理。
当队列为空且生产者已完成时,消费者停止。
"""
global i # 用于计数处理的句子
while True:
# 检查是否应该停止:队列为空且生产者已完成
if q.empty() and flag.is_set():
break
try:
# 尝试从队列获取项,如果队列为空,会等待
item = await asyncio.wait_for(q.get(), timeout=1.0) # 增加超时,避免无限等待
except asyncio.TimeoutError:
# 如果超时且生产者已完成,则退出
if flag.is_set():
break
continue # 否则继续等待
print("processing sentence: ", i)
print("waiting for processing sentence: ", item)
await asyncio.sleep(len(item) * 0.1)
print("sentence processed!")
i += 1
async def main():
global i
i = 1 # 初始化句子计数器
event = asyncio.Event() # 用于生产者通知消费者完成
queue = asyncio.Queue[str]() # 共享队列
# 创建生产者和消费者任务
producer_task = sentences_generator(queue, event)
consumer_task = process_sentence(queue, event)
# 并发运行生产者和消费者任务
await asyncio.gather(producer_task, consumer_task)
asyncio.run(main())代码解析:
sentences_generator (生产者):
process_sentence (消费者):
main 函数:
预期输出(部分):
got char: H got char: i got char: . got sentence: Hi. got char: got char: H got char: e got char: l got char: l got char: o got char: . got sentence: Hello. processing sentence: 1 waiting for processing sentence: Hi. got char: got char: T got char: h got char: a got char: n got char: k got char: got char: y got char: o got char: u got char: . got sentence: Thank you. sentence processed! processing sentence: 2 waiting for processing sentence: Hello. sentence processed! processing sentence: 3 waiting for processing sentence: Thank you. sentence processed!
从新的输出中可以看到,当process_sentence正在处理"Hi."时,stream和sentences_generator已经继续生成了"Hello."甚至"Thank you."。这种交错的输出表明生产者和消费者正在并发地工作,显著提高了程序的效率和响应性。
通过采用生产者-消费者模式并结合asyncio.Queue和asyncio.Event,我们可以有效地管理异步任务间的依赖关系,实现更高效、更具响应性的并发数据流处理。这对于构建复杂的异步系统,如网络服务、数据管道等,是至关重要的技术。
以上就是优化Asyncio嵌套函数调度:使用生产者-消费者模式实现并发流处理的详细内容,更多请关注其它相关文章!
# 我们可以
# 石家庄网站搜索优化服务
# seo优化排名官方版
# 浦东seo优化报价
# 新乡搜狗关键词排名技术
# 投资公司网站建设流程
# 上海营销推广企业排名榜
# seo实战进展
# 罗湖手机网站优化效果好
# 宿松网站建设制作外包
# 卖车位营销推广关键词
# 的是
# 几种
# 并将
# python
# 它是
# 会在
# 重构
# 浮点
# 多个
# 为空
# 异步任务
# stream
# ai
# 工具
# 处理器
# go
相关栏目:
【
Google疑问12 】
【
Facebook疑问10 】
【
优化推广96088 】
【
技术知识133117 】
【
IDC资讯59369 】
【
网络运营7196 】
【
IT资讯61894 】
相关推荐:
Teambition网盘如何共享文件
Go语言中方法与接收器:指针和值类型的调用机制详解
AffinityDesigner图层蒙版怎么用_AffinityDesigner图层蒙版设计应用
Sublime怎么格式化HTML代码_Sublime前端代码美化插件使用指南
汽水音乐官网网页版入口 汽水音乐官网网页版在线入口
Bootstrap 5导航栏折叠功能失效:数据属性迁移指南
优化 WooCommerce 产品价格显示与自定义短代码集成
键盘测试软件哪个好_键盘故障检测工具推荐
《淘宝联盟》推广自己的店铺方法
批改网网页版登录 批改网电脑版学生登录入口
花生壳内网映射新方案
realme 10 Pro息屏方案_realme 10 Pro省电策略
抖音如何解除|直播|权限绑定_抖音关闭并解绑|直播|功能的方法
使用CSS :has() 选择器实现父元素样式控制:从子元素反向应用样式
实现可重用自定义Python Range类
如何外贸网站设计-能留住客户提升用户体验!
优化CSS动画与J*aScript定时器协同:构建稳定Toast提示
创客贴登录页面入口 创客贴网页版最新网址链接
WPS长文档分栏排版不乱方法_WPS分栏+分节符报纸排版教程
PSD转AI文件的简单方法
TikTok搜索结果不显示怎么办 TikTok搜索刷新与优化方法
J*aScript类型数组_TypedArray使用
一加 Ace 6V 快充无法启用_一加 Ace 6V 充电优化
PHP utf8_encode 字符编码转换陷阱与解决方案
如何在Golang中处理表单文件上传_Golang 表单文件上传示例
J*aScript模块加载器_RequireJS原理分析
金牛福袋获取攻略
智学网成绩单查询系统网_智学网学生平台登录
Win10关闭UAC用户账户控制的方法 Win10降低安全提示等级【技巧】
msn官方入口2025登录 msn官网2025直达首页入口
《虎扑》关闭社区内容推荐方法
斯宾塞称XGP云游戏“蒸蒸日上”:正在构建一个游戏从未如此唾手可得的未来
PHP与SQL实践:高效实现数据复制与特定列值修改
word文档行距怎么调?word文档调行距的操作步骤
139邮箱登录入口官网 139邮箱登录入口官网网址
12306不能订票的时间段是固定的吗? | 节假日购票时间有无变化
《知到》打卡课程方法
教育查询官方网站入口 教育个人档案查询免费官网
J*a中导出MySQL表为SQL脚本的两种方法
Safari浏览器自动填表功能失效怎么办 Safari表单管理修复
个人所得税办理入口 个人所得税综合所得年度汇算入口
漫蛙manwa官网浏览入口_漫蛙漫画网页版访问链接
抖音火山版注销账号抖音会注销吗 抖音火山版与抖音账号注销关系
Django模型动态关联检查:高效管理复杂关系
优化Google Charts Gauge:在数据库无数据时显示默认值
iPhone 15 Pro如何查看存储空间占用_iPhone 15 Pro存储空间查看教程
解决Windows上Composer PATH变量冲突导致的命令无法识别问题
Mac怎么关闭按键声音_Mac键盘打字音效设置
《随手记》启用语音备注方法
抖音赚钱快速入门_新手必看的抖音赚钱步骤
2025-11-29
运城市盐湖区信雨科技有限公司是一家深耕海外推广领域十年的专业服务商,作为谷歌推广与Facebook广告全球合作伙伴,聚焦外贸企业出海痛点,以数字化营销为核心,提供一站式海外营销解决方案。公司凭借十年行业沉淀与平台官方资源加持,打破传统外贸获客壁垒,助力企业高效开拓全球市场,成为中小企业出海的可靠合作伙伴。