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

python中实现协程需使用async def定义协程函数，await暂停执行并交出控制权，事件循环通过asyncio.run()启动，实现单线程内高效并发I/O操作。

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

等待它们完成。你会发现，虽然每个任务都有自己的等待时间，但总的执行时间远小于它们等待时间的总和，因为它们是并发执行的。

立即学习“Python免费学习笔记（深入）”；

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密集型或简单任务中，它可能只是徒增烦恼。

Shrink.media

Shrink.media是当今市场上最快、最直观、最智能的图像文件缩减工具

83

查看详情

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

   的对象完成。理解这一点能帮你更好地控制任务的生命周期和依赖关系。

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