Python asyncio异步编程五大常见陷阱与避坑指南

asyncio 堪称现代 Python 开发的必备技能，也是面试中的高频考点。然而，许多开发者在实际使用时，常常会发现自己写的“异步”代码并没有带来预期的性能提升，甚至会出现意料之外的阻塞或错误。你是否也遇到过这些困惑：

• 代码明明用了 async def，跑起来却比同步还慢？
• 已经写了 await，程序为何还是“卡死”了？
• 使用的 await 越来越多，性能反而越来越差？

本文将带你剖析五个在 Python 异步编程中极易踩入的“坑”，并提供清晰的解决方案，帮助你写出真正高效、正确的异步代码。

坑1：同步阻塞代码混入异步函数

这是最经典且最常见的问题。

典型错误示例：

import asyncio
import time

async def fetch_data():
    print("开始获取数据...")
    time.sleep(2)  # ❌ 这里是同步阻塞！
    return "数据拿到"

async def main():
    result = await fetch_data()
    print(result)

asyncio.run(main())

问题所在：time.sleep(2) 是一个同步的阻塞调用，它会挂起整个线程，导致事件循环也被完全阻塞。此时，其他所有异步任务都无法得到执行机会，异步的优势荡然无存。

正确做法：

import asyncio

async def fetch_data():
    print("开始获取数据...")
    await asyncio.sleep(2)  # ✅ 使用异步睡眠，让出控制权
    return "数据拿到"

async def main():
    result = await fetch_data()
    print(result)

asyncio.run(main())

核心原则：在 async 函数内部，永远不要使用 time.sleep()、requests.get()（未使用异步客户端）这类同步阻塞操作，应替换为其对应的异步版本（如 asyncio.sleep()、aiohttp）。

坑2：忘记使用 await 关键字

忘记了 await，等于没有真正执行异步函数。

典型错误示例：

import asyncio

async def get_data():
    await asyncio.sleep(1)
    return "666"

async def main():
    result = get_data()  # ❌ 没有 await！
    print(result)        # 输出：<coroutine object get_data at 0x...>

asyncio.run(main())

问题所在：直接调用一个异步函数（协程）并不会执行它，而是返回一个协程对象（coroutine object）。你需要使用 await 来等待并获取其执行结果。

正确做法：

async def main():
    result = await get_data()  # ✅ 加上 await
    print(result)              # 输出：666

血的教训：忘记 await 意味着你的异步任务只是在事件循环中排队，从未被真正触发执行。

坑3：以串行方式编写并发任务

这是导致异步代码性能不升反降的常见原因。

典型错误示例：

import asyncio

async def task(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 结束")

async def main():
    await task("任务1", 2) # ❌ 串行执行
    await task("任务2", 2)
    await task("任务3", 2)

# 总耗时：6秒

问题所在：使用连续的 await 会导致任务一个接一个地执行，总耗时是所有任务耗时的总和，这完全失去了并发执行的意义。

正确做法：
使用 asyncio.create_task() 创建任务，然后使用 asyncio.gather() 等待它们完成。

async def main():
    # ✅ 创建任务，它们会并发执行
    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秒

更简洁的写法：

async def main():
    await asyncio.gather(
        task("任务1", 2),
        task("任务2", 2),
        task("任务3", 2),
    )

核心机制：asyncio.gather() 或 asyncio.create_task() 能够将多个协程调度为并发执行，当某个任务在 await（如 asyncio.sleep）时，事件循环会切换到其他就绪任务，从而实现高效的 高并发 处理。

坑4：在同步函数中直接调用异步函数

异步世界和同步世界不能直接互通。

典型错误示例：

import asyncio

async def async_func():
    await asyncio.sleep(1)
    return "ok"

def sync_func():
    result = async_func()  # ❌ 同步函数无法直接调用/等待异步函数
    return result

# 运行会报错：RuntimeError/TypeError

问题所在：同步函数没有运行在事件循环中，无法理解和等待一个协程对象。

解决方案：
如果你必须在同步上下文中获取异步函数的结果，需要显式地运行一个事件循环。

# 方案1：使用 asyncio.run() （Python 3.7+）
def sync_func():
    return asyncio.run(async_func())

# 方案2：手动管理事件循环
def sync_func():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        result = loop.run_until_complete(async_func())
    finally:
        loop.close()
    return result

重要原则：异步调用链应尽可能保持“全异步”，从入口到最底层。如果必须混合，需在边界处妥善处理事件循环。

坑5：嵌套使用 asyncio.run()

这是一个典型的运行时错误。

典型错误示例：

import asyncio

async def inner():
    print("inner 开始")
    await asyncio.sleep(1)
    print("inner 结束")
    return "inner result"

async def outer():
    print("outer 开始")
    result = asyncio.run(inner())  # ❌ 在已运行的事件循环中再次调用 asyncio.run()
    print(f"outer 结束, 结果: {result}")

asyncio.run(outer())

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

问题所在：asyncio.run() 是一个高级封装函数，它会创建新的事件循环并运行传入的协程，运行完毕后关闭循环。在已经运行着事件循环的协程内部再次调用它，会导致冲突。

正确做法：
在异步函数内部调用另一个异步函数，直接使用 await 即可。

async def outer():
    print("outer 开始")
    result = await inner()  # ✅ 直接 await
    print(f"outer 结束, 结果: {result}")

asyncio.run(outer())

核心要点：一个线程中通常只有一个主事件循环。程序的异步入口使用一次 asyncio.run()，其内部的所有异步调用都应通过 await 链式进行。

总结回顾

为了方便记忆，我们将上述五个常见陷阱及其解决方法总结如下：

坑点	典型症状	解决方案
混用同步阻塞代码	事件循环被阻塞，异步失效	使用对应的异步库和函数（如 asyncio.sleep）
忘记 await	得到协程对象而非结果，函数未执行	在调用异步函数时加上 await 关键字
串行写并发	多个“异步”任务总耗时极长	使用 asyncio.gather() 或 asyncio.create_task()
同步函数调用异步函数	直接调用报 RuntimeError/TypeError	在边界使用 asyncio.run() 或手动管理事件循环
嵌套 asyncio.run()	报 RuntimeError	在异步函数内部直接 await 其他异步函数

asyncio 的核心思想在于 避免阻塞事件循环，充分利用 I/O 等待时间让其他任务交叉执行。理解并规避上述陷阱，是掌握 Python 异步编程的关键一步。希望这篇指南能帮助你在 云栈社区 的编程之路上走得更稳。在实践中如果遇到其他有趣的“坑”，也欢迎分享与探讨。