Python协程中未await的async函数直接报RuntimeWarning

当你在编写异步代码时，如果控制台突然弹出一条 RuntimeWarning: coroutine 'xxx' was never awaited，这意味着你的程序中存在一个潜在的严重错误。本文将直接解释这个警告的含义、产生的原因，并提供清晰、可执行的修复步骤。

理解错误信息的含义

这个警告的核心含义是：你的代码中创建了一个异步任务（协程对象），但没有驱动它执行。

• “coroutine ‘xxx’”：这里的 xxx 是你的异步函数名，例如 my_async_func。它表明这个函数本身是一个“协程工厂”，调用它会返回一个“协程对象”。
• “was never awaited”：await 关键字的作用是“暂停当前协程的执行，直到被等待的协程完成”。如果你从未 await 这个协程对象，它就永远不会被安排到事件循环中执行，变成了一段“从未运行的代码”。

展示一个典型的错误代码示例

请观察下面这段有问题的代码，它试图异步获取用户信息和用户订单。

import asyncio

async def fetch_user_data(user_id):
    print(f“正在获取用户 {user_id} 的数据...”)
    await asyncio.sleep(1) # 模拟网络请求
    return {“id”: user_id, “name”: “张三”}

async def fetch_user_orders(user_id):
    print(f“正在获取用户 {user_id} 的订单...”)
    await asyncio.sleep(1.5) # 模拟网络请求
    return [“订单A”, “订单B”]

async def main():
    user_id = 101
    # 错误：下面两行调用 async 函数时，没有使用 await
    user_data = fetch_user_data(user_id)
    orders = fetch_user_orders(user_id)

    print(“用户数据:”, user_data)
    print(“用户订单:”, orders)

if __name__ == “__main__”:
    asyncio.run(main())

运行此代码，你将立即看到两个 RuntimeWarning，并且 user_data 和 orders 的值不是我们期望的字典和列表，而是类似 <coroutine object fetch_user_data at 0x...> 的协程对象。

分析错误的根本原因

在 main 函数中，fetch_user_data(user_id) 和 fetch_user_orders(user_id) 这两行代码：

1. 确实被调用了：Python 解释器执行了 fetch_user_data 函数体内的代码（直到第一个 await 语句 asyncio.sleep(1)），但并没有完成它。
2. 但返回的是“任务描述”：async 函数被调用后，不会像普通函数一样返回结果。它立即返回一个“协程对象”。这个对象包含了执行该函数所需的所有信息和状态。
3. 对象被丢弃：我们将这个协程对象赋值给了变量（如 user_data），但随后只打印了这个对象本身，并没有使用 await 去驱动它运行至完成。事件循环从未获得机会去执行它。

简单比喻：你订购了一个汉堡（调用 async 函数），得到了一个取餐号（协程对象）。但你从未拿着取餐号去取餐（await），所以汉堡永远不会送到你手上。

分步解决方案

要消除 RuntimeWarning 并让异步代码正确工作，你必须确保每一个协程对象最终都被 await。

1. 理解await的用法

await 只能用于被 await 的上下文中，即另一个 async 函数内部，或直接传递给事件循环。

• 在async函数内：这是最常用的场景。你可以在 async 函数体内使用 await 来等待其他 async 函数完成。
• 使用asyncio.run()启动：asyncio.run() 函数是启动顶层 async 函数（通常命名为 main）的推荐方式。它内部负责创建事件循环和驱动协程。

2. 修复上文的错误代码

根据错误原因，对 main 函数进行如下修改：

async def main():
    user_id = 101
    # 正确：使用 await 来驱动协程执行并获取结果
    user_data = await fetch_user_data(user_id)
    orders = await fetch_user_orders(user_id)

    print(“用户数据:”, user_data)
    print(“用户订单:”, orders)

3. 验证修复效果

运行修复后的完整代码：

import asyncio

async def fetch_user_data(user_id):
    print(f“正在获取用户 {user_id} 的数据...”)
    await asyncio.sleep(1)
    return {“id”: user_id, “name”: “张三”}

async def fetch_user_orders(user_id):
    print(f“正在获取用户 {user_id} 的订单...”)
    await asyncio.sleep(1.5)
    return [“订单A”, “订单B”]

async def main():
    user_id = 101
    user_data = await fetch_user_data(user_id)
    orders = await fetch_user_orders(user_id)

    print(“用户数据:”, user_data)
    print(“用户订单:”, orders)

if __name__ == “__main__”:
    asyncio.run(main())

现在，控制台将输出预期的执行流程和结果，不再有 RuntimeWarning。两个网络请求将顺序执行（总计约2.5秒）。

4. 掌握更高级的并发模式

如果你希望 fetch_user_data 和 fetch_user_orders 并发执行以节省时间，可以使用 asyncio.gather 或 asyncio.create_task。这同样需要正确处理协程对象。

async def main_concurrent():
    user_id = 101
    # 创建两个协程对象
    coro_user = fetch_user_data(user_id)
    coro_orders = fetch_user_orders(user_id)

    # 使用 gather 来并发地 await 多个协程
    user_data, orders = await asyncio.gather(coro_user, coro_orders)

    print(“用户数据:”, user_data)
    print(“用户订单:”, orders)

或者使用 create_task 创建明确的任务：

async def main_with_tasks():
    user_id = 101
    # 创建任务，并立即安排它们到事件循环中执行
    task_user = asyncio.create_task(fetch_user_data(user_id))
    task_orders = asyncio.create_task(fetch_user_orders(user_id))

    # 等待所有任务完成
    user_data = await task_user
    orders = await task_orders

    print(“用户数据:”, user_data)
    print(“用户订单:”, orders)

5. 检查常见的遗漏场景

请检查你的代码中是否在以下位置错误地调用了 async 函数而未 await：

• 在列表推导式或生成器表达式中。
• 作为其他函数的普通参数传递时（除非该函数本身设计用来接受协程，如 asyncio.gather）。
• 在 try...except 块的 finally 子句中忘记 await 清理资源。

如何主动避免此类错误

1. 启用警告：确保 ResourceWarning 没有被过滤。在开发环境中，可以运行 python -W error your_script.py，这会把所有警告当作错误抛出，强制你解决问题。
2. 使用类型提示：为函数参数和返回值添加类型提示（如 async def func() -> int:），可以在某些编辑器和类型检查器中提前获得关于未使用 await 的反馈。
3. 代码审查：养成习惯，每当你写下一行调用 async 函数的代码时，都自问：“我是否需要立即得到它的结果？” 如果需要，就必须 await。如果只是想将其注册为后台任务，则使用 asyncio.create_task()。