python中怎么实现协程（async/await）？（python.async.await...）

wufei123 发布于 2025-09-17 阅读（9）
Python中实现协程需使用async def定义协程函数，await暂停执行并交出控制权，事件循环通过asyncio.run()启动，实现单线程内高效并发I/O操作。

python中怎么实现协程（async/await）？

Python中实现协程，核心就是利用 

async
和 
await
这两个关键字，它们是 
asyncio
库提供的高级抽象。简单来说，协程让程序可以在等待某些操作（比如网络请求、文件读写）完成时，暂停当前任务，去执行其他任务，从而提高效率，尤其是在I/O密集型应用中。它不是真正的并行，而是一种协作式多任务，让你的代码能够“一边等水烧开，一边切菜”。 解决方案

第一次接触 

async/await
的时候，可能会觉得有点绕，但一旦你掌握了它的基本模式，就会发现它其实非常直观且强大。实现协程，你主要需要关注以下几点：
  1. 定义协程函数： 使用 
    async def
    关键字来定义一个协程函数。这样的函数在被调用时不会立即执行，而是返回一个协程对象。
  2. 等待协程完成： 在另一个协程函数内部，使用 
    await
    关键字来暂停当前协程的执行，直到 
    await
    后面的协程（或可等待对象，如 
    asyncio.sleep
    ）完成。
  3. 运行事件循环： 协程的执行需要一个事件循环（Event Loop）。
    asyncio.run()
    是运行顶层协程最简单的方式，它会负责创建和管理事件循环。

下面是一个简单的例子，展示了如何定义和运行协程：

import asyncio
import time

async def fetch_data(item_id, delay):
    """
    模拟一个异步的网络请求或数据库查询。
    它会等待一段时间，然后返回一些数据。
    """
    print(f"[{time.strftime('%H:%M:%S')}] 正在为 {item_id} 模拟数据获取，预计等待 {delay} 秒...")
    await asyncio.sleep(delay) # 这是一个可等待对象，会暂停当前协程，但不阻塞事件循环
    print(f"[{time.strftime('%H:%M:%S')}] {item_id} 数据获取完成。")
    return f"数据 for {item_id}"

async def process_items():
    """
    主协程，负责创建并运行多个数据获取任务。
    """
    print(f"[{time.strftime('%H:%M:%S')}] 开始处理所有项目...")

    # 创建多个协程任务，它们将并发运行
    task1 = asyncio.create_task(fetch_data("商品A", 3))
    task2 = asyncio.create_task(fetch_data("商品B", 1))
    task3 = asyncio.create_task(fetch_data("商品C", 2))

    # 等待所有任务完成。await 会暂停当前协程，直到所有指定的任务都完成。
    # 这里也可以使用 asyncio.gather() 来更优雅地等待多个任务。
    result_a = await task1
    result_b = await task2
    result_c = await task3

    print(f"[{time.strftime('%H:%M:%S')}] 所有项目处理完成。")
    print(f"获取到的数据: {result_a}, {result_b}, {result_c}")

if __name__ == "__main__":
    # 运行主协程。asyncio.run() 会自动管理事件循环的启动和关闭。
    start_time = time.perf_counter()
    asyncio.run(process_items())
    end_time = time.perf_counter()
    print(f"\n总耗时: {end_time - start_time:.2f} 秒")

在这个例子中，

fetch_data
是一个协程函数，它模拟了一个耗时的I/O操作。
process_items
是另一个协程，它通过 
asyncio.create_task
创建了三个 
fetch_data
任务，并用 
await
等待它们完成。你会发现，虽然每个任务都有自己的等待时间，但总的执行时间远小于它们等待时间的总和，因为它们是并发执行的。 async/await 背后的原理是什么？它和多线程有什么区别？

很多人刚开始都会把协程和多线程搞混，这很正常。但它们俩的哲学完全不同。要理解 

async/await
，核心在于理解事件循环（Event Loop）和协作式多任务。

当一个协程遇到 

await
关键字时，它会暂停自己的执行，将控制权交还给事件循环。事件循环此时会检查是否有其他协程已经准备好运行（比如，之前等待的网络请求已经返回数据了）。如果有，事件循环就会去运行那个准备好的协程。当之前 
await
的操作完成（比如数据真的从网络回来了），事件循环会再次调度原先暂停的协程，让它从 
await
的地方继续执行。整个过程都在一个单线程中完成。

这和多线程的抢占式多任务形成了鲜明对比：

  • 多线程： 操作系统是老大，它决定什么时候暂停一个线程，什么时候运行另一个线程，线程之间是“抢”CPU资源的。这意味着线程切换（上下文切换）的开销相对较大，而且在Python中，由于全局解释器锁（GIL）的存在，即便在多核CPU上，Python解释器在同一时刻也只能执行一个线程的字节码，这限制了CPU密集型任务的并行度。多线程更像是你雇了几个工人，让他们各自去完成不同的任务，但他们之间可能会抢工具，也需要协调。
  • 协程（async/await）： 协程是“君子协定”，它们自己决定什么时候暂停，什么时候让出控制权。它们在一个线程里“轮流”执行，没有GIL的限制（因为只有一个线程），上下文切换的开销非常小。协程更像是你一个人，但你学会了高效地切换任务：水烧上了就去切菜，菜切好了就去洗米，水开了就去煮饭。这种方式特别适合I/O密集型任务，因为等待I/O的时间可以用来做其他事情，而不是白白浪费CPU周期。

所以，核心区别在于：多线程是操作系统级别的并行或并发，适用于CPU密集型任务（如果能绕开GIL的话）和I/O密集型任务；而协程是应用程序级别的并发，非常适合I/O密集型任务，因为它能以极低的开销实现大量并发连接，但对CPU密集型任务无能为力（因为仍然是单线程）。

在实际项目中，什么时候应该考虑使用 async/await？

这其实是个很实际的问题，不是所有场景都适合 

async/await
。我个人觉得，它最闪耀的舞台，通常是那些需要处理大量并发I/O操作的场景。

你可以这样思考：

  • 当你需要等待外部资源时： 比如网络请求（调用API、爬虫）、数据库查询、文件读写、消息队列的消费与生产等。这些操作通常涉及等待数据从慢速设备（网络、磁盘）传输，等待期间CPU是空闲的。
    async/await
    能让你的程序在这段空闲时间里去处理其他任务，而不是傻等。
    • 例子： 构建高性能的Web服务（如基于FastAPI或Aiohttp）、并发抓取大量网页数据、处理大量实时消息流。
  • 当你需要构建响应式、高吞吐量的服务时： 如果你的服务需要同时处理成千上万个客户端连接，并且每个连接都有一定的等待时间（比如聊天服务器、游戏后端），
    asyncio
    就能发挥巨大作用，因为它能用一个（或少量）线程处理大量并发连接，大大节省了系统资源。
    • 例子： WebSocket服务器、长连接服务。
  • 当你发现传统的多线程/多进程模型开销太大或管理复杂时： 线程或进程的创建和销毁、上下文切换都有不小的开销，而且多线程的同步（锁、信号量）问题也容易引入bug。协程则轻量得多，且避免了复杂的共享内存同步问题（因为通常在一个线程内）。

什么时候不适合呢？

  • CPU密集型任务： 如果你的任务主要是进行大量计算，比如图像处理、复杂算法、数据分析中的聚合操作等，
    async/await
    帮不了你。因为它还是单线程，计算密集型任务会直接阻塞整个事件循环。这种情况下，多进程（
    multiprocessing
    ）才是更好的选择，它可以利用多核CPU进行真正的并行计算。
  • 简单的脚本或没有I/O等待的任务： 如果你只是想写个脚本，读取几个本地文件，然后做一些纯粹的计算，那可能 
    asyncio
    反而会把事情搞复杂。引入 
    async/await
    会增加代码的复杂性，如果收益不明显，就没有必要。
  • 现有的同步代码库： 如果你大量依赖的第三方库是同步的（比如 
    requests
    库），那么直接在 
    async
    函数里调用它们会阻塞事件循环。虽然有 
    asyncio.to_thread()
    可以把同步阻塞代码放到单独的线程池里运行，但这也增加了额外的复杂性。

总而言之，

async/await
是一个强大的工具，但它有自己的适用范围。在I/O密集型应用中，它能让你事半功倍；但在CPU密集型或简单任务中，它可能只是徒增烦恼。 Post AI Post AI

博客文章AI生成器

Post AI50 查看详情 Post AI async/await 使用中常见的坑和调试技巧有哪些？

在 

async/await
的世界里遨游，虽然很爽，但偶尔也会踩到一些“暗礁”。我个人踩过最大的坑，就是在 
async def
函数里，不小心调用了一个同步的阻塞I/O函数，结果整个事件循环都被卡住了，服务直接“假死”。

常见的坑：

  1. 在 
    async
    函数中调用阻塞的同步代码： 这是最致命的错误之一。比如在 
    async def
    函数里直接用 
    time.sleep()
    而不是 
    asyncio.sleep()
    ，或者使用 
    requests.get()
    而不是 
    aiohttp
    等异步HTTP库。这会直接阻塞事件循环，让你的并发优势荡然无存。
    • 解决方法： 尽量使用异步版本的库。如果实在需要调用同步阻塞代码，可以考虑使用 
      await asyncio.to_thread(sync_blocking_function, *args)
      ，它会在一个单独的线程中运行同步函数，避免阻塞主事件循环。
  2. 忘记 
    await
    一个协程对象： 当你调用一个 
    async def
    函数时，它返回的是一个协程对象，而不是执行结果。如果你忘记 
    await
    它，这个协程就不会被调度执行，也不会报错（或者只在程序退出时报一个 
    RuntimeWarning: coroutine '...' was never awaited
    ）。
    • 示例： 
      my_coroutine()
      而不是 
      await my_coroutine()
      或 
      asyncio.create_task(my_coroutine())
      。
    • 解决方法： 确保每个协程对象都被 
      await
      了，或者被 
      asyncio.create_task()
      调度成一个任务。
  3. 不正确地处理任务异常： 如果一个通过 
    asyncio.create_task()
    创建的任务抛出异常，但你没有 
    await
    这个任务，或者没有在 
    await
    时捕获异常，那么这个异常可能不会立即被发现，甚至可能导致整个事件循环崩溃。
    • 解决方法： 始终 
      await
      你的任务，并在 
      await
      语句周围使用 
      try...except
      。或者，给任务添加一个 
      done_callback
      来处理异常。
      asyncio.gather()
      在等待多个任务时，会将所有异常都收集起来。
  4. asyncio.run()
    的限制： 
    asyncio.run()
    只能在主线程中调用一次，且不能嵌套。如果你在一个已经运行的事件循环中尝试再次调用它，会报错。
    • 解决方法： 如果你已经在事件循环中，并且需要运行另一个协程，可以直接使用 
      await
      或 
      asyncio.create_task()
      。

调试技巧：

  1. 开启 

    asyncio
    调试模式： 这是我调试 
    asyncio
    代码的首选。你可以通过 
    python -m asyncio your_script.py
    来运行脚本，或者在代码中设置 
    loop.set_debug(True)
    。调试模式会提供更详细的日志输出，比如未被 
    await
    的协程、阻塞事件循环的警告等。
    import asyncio

async def main():
    # ... your async code ...
    pass

if __name__ == "__main__":
    # 获取当前事件循环并设置调试模式
    loop = asyncio.get_event_loop()
    loop.set_debug(True)
    # 运行主协程
    loop.run_until_complete(main())
    loop.close()

    （注意：

    asyncio.run()
    内部已经包含了事件循环的创建和关闭，如果你用 
    asyncio.run()
    ，则无法直接设置 
    loop.set_debug(True)
    。但 
    asyncio.run()
    也可以通过 
    debug=True
    参数来开启调试模式：
    asyncio.run(main(), debug=True)
    。）

  2. 日志记录： 在关键的协程入口和出口添加日志，记录任务的开始、结束、以及任何重要状态。这有助于追踪任务的执行流程和发现异常。

  3. 使用 

    asyncio.all_tasks()
    ： 在调试时，你可以用 
    asyncio.all_tasks()
    来查看当前事件循环中所有正在运行或等待的任务，这能帮你了解事件循环的“繁忙”程度和任务状态。

  4. 逐步调试器： 虽然调试异步代码比同步代码复杂一些，但像 VS Code 这样的IDE提供的调试器仍然非常有用。学会如何在 

    await
    语句处设置断点，以及如何步入（step into）和步过（step over）协程调用，是提高效率的关键。

  5. 理解 

    asyncio.create_task()
    和 
    await
    的区别： 
    create_task()
    是将一个协程包装成一个 
    Task
    对象，并提交给事件循环，它会立即返回一个 
    Task
    对象，协程会在后台运行。而 
    await
    则是等待一个可等待对象完成，它会暂停当前协程直到被 
    await
    的对象完成。理解这一点能帮你更好地控制任务的生命周期和依赖关系。

调试异步代码确实需要一些耐心和经验，但通过利用这些工具和理解其工作原理，你将能更有效地定位和解决问题。

以上就是python中怎么实现协程（async/await）？的详细内容，更多请关注知识资源分享宝库其它相关文章！

相关标签： python 操作系统 websocket 工具 后端 ai 爬虫 解决方法 区别 Python fastapi try 循环 Event 线程 多线程 主线程 并发 对象 事件 异步 ide 算法 数据库 数据分析 http websocket bug 大家都在看： Python字符串多重替换：解决迭代更新的常见陷阱 Python中列表字面量、range与迭代器内存行为深度解析 使用Python将LineString转换为带缓冲区的多边形 Python列表推导式与迭代器内存行为深度解析 Python列表字面量、迭代器与内存管理：深度解析即时求值行为

标签:  python async await 

相关阅读

发表评论:

◎欢迎参与讨论，请在这里发表您的看法、交流您的观点。

最近发表

标签列表