当前位置:首页>python>Python 异步编程 asyncio 的那些坑

Python 异步编程 asyncio 的那些坑

  • 2026-09-02 20:17:20
Python 异步编程 asyncio 的那些坑

❌ 你的 async 代码真的异步了吗?⚡ 5个常见翻车现场,最后一个我踩了3次!


📖 引言

asyncio 应该是 Python 面试必问知识点,也是现代 Python 开发的必备技能。

但!很多同学写着写着就出了问题:

  • • 代码看起来是 async,结果跑起来比同步还慢?
  • • 明明写了 async def,为啥还卡死?
  • • 用的 await 越来越多,性能反而越来越拉?

这篇文章,就是帮你避坑的。


🔍 坑1:同步代码混进异步世界

典型症状

import asyncioimport timeasyncdeffetch_data():print("开始获取数据...")    time.sleep(2)  # ❌ 这是同步阻塞!return"数据拿到"asyncdefmain():    result = await fetch_data()print(result)asyncio.run(main())

问题:time.sleep(2) 会阻塞整个事件循环!这TM比同步还同步。

正确姿势

import asyncioasyncdeffetch_data():print("开始获取数据...")await asyncio.sleep(2)  # ✅ 异步睡眠,不阻塞return"数据拿到"asyncdefmain():    result = await fetch_data()print(result)asyncio.run(main())

💡 记住:在 async 函数里,永远不要用time.sleep(),用 asyncio.sleep()


🔍 坑2:忘记 await = 拿到一个"假结果"

典型症状

import asyncioasyncdefget_data():await asyncio.sleep(1)return"666"asyncdefmain():    result = get_data()  # ❌ 没 await!print(result)        # 打印的是 <coroutine object...>print(result)        # 还是 <coroutine object...>asyncio.run(main())

正确姿势

asyncdefmain():    result = await get_data()  # ✅ 加 awaitprint(result)              # 打印 666

⚠️ 血的教训:忘记 await = 你的函数根本没执行,只是在排队


🔍 坑3:并发任务不会"一起跑"

典型症状

import asyncioasyncdeftask(name, delay):print(f"{name} 开始")await asyncio.sleep(delay)print(f"{name} 结束")asyncdefmain():await task("任务1", 2)await task("任务2", 2)await task("任务3", 2)# 总耗时:6秒 ❌

问题:你TM写的是串行!3个任务一个接一个,6秒是必然的。

正确姿势

asyncdefmain():# ✅ 创建任务,不阻塞    t1 = asyncio.create_task(task("任务1", 2))    t2 = asyncio.create_task(task("任务2", 2))    t3 = asyncio.create_task(task("任务3", 2))# 等待所有任务完成await asyncio.gather(t1, t2, t3)# 总耗时:2秒 ✅

或者更简洁:

asyncdefmain():await asyncio.gather(        task("任务1", 2),        task("任务2", 2),        task("任务3", 2),    )

🚀 真相:asyncio.gather = 并发执行多个协程,性能直接翻倍


🔍 坑4:在同步函数里调用 async 函数

典型症状

import asyncioasyncdefasync_func():await asyncio.sleep(1)return"ok"defsync_func():    result = async_func()  # ❌ 同步函数不能直接调asyncreturn result# 运行直接报错:TypeError: An asyncio.Future...

解决方案

# 方案1:在入口处用 asyncio.run()defsync_func():return asyncio.run(async_func())# 方案2:如果在已有事件循环中,用 loop.run_until_complete()defsync_func():    loop = asyncio.new_event_loop()    result = loop.run_until_complete(async_func())    loop.close()return result

🎯 原则:async 调用链必须从头到尾都是 async,一断全断


🔍 坑5:asyncio.run() 嵌套 - 经典作死

典型症状

import asyncioasyncdefinner():print("inner 开始")await asyncio.sleep(1)print("inner 结束")return"inner result"asyncdefouter():print("outer 开始")    result = asyncio.run(inner())  # ❌ 在 async 里又开 asyncio.run()print(f"outer 结束, 结果: {result}")asyncio.run(outer())

报错:RuntimeError: asyncio.run() cannot be called from a running event loop

正确姿势

asyncdefouter():print("outer 开始")    result = await inner()  # ✅ 直接 awaitprint(f"outer 结束, 结果: {result}")asyncio.run(outer())

🔥 记住:入口用 asyncio.run(),内部用 await


📌 总结

坑点
症状
解药
用 time.sleep
事件循环阻塞
用 asyncio.sleep
忘 await
返回协程对象
加 await
串行写并发
性能差
用 asyncio.gather/create_task
同步调异步
报错
保持调用链全异步
asyncio.run() 嵌套
运行时错误
内部用 await

asyncio 的核心就一句话:别阻塞事件循环,让任务交叉执行。


👋 行动引导

看到这里,说明你对 asyncio 有追求!

点赞 👍 + 收藏 ⭐ + 关注 🔔

我是小甲鱼,每周更新硬核 Python 技术。下一期想看什么?评论区告诉我!

🐟 跟着小甲鱼,学 Python 不迷路

最新文章

随机文章