python协程与asyncio通过协作式并发高效处理I/O密集任务，相比多线程/多进程，其在单线程内以await暂停协程，由事件循环调度，避免GIL限制与线程切换开销，适用于爬虫、异步Web服务、数据库操作等场景，并通过asyncio.create_task、gather和异常处理机制实现任务管理与健壮性控制。
Python协程和asyncio，在我看来，是Python处理并发I/O操作的一套优雅且高效的机制，它让单线程程序也能“同时”处理多项任务，而无需承担多线程或多进程带来的复杂性和开销。简单来说，协程是一种可暂停和恢复的函数，而asyncio则是Python内置的事件循环框架，负责调度和管理这些协程的执行，让它们在等待I/O时能够“让出”CPU，从而提高程序的吞吐量。它不是为了榨干CPU的多核性能，而是为了更有效地利用等待I/O的时间。
解决方案
要深入理解Python协程和asyncio，我们得从它们各自的角色说起。协程，本质上是一种用户态的轻量级线程，或者说是一种特殊的生成器函数。在Python中，我们通过
async def
来定义一个协程函数，而
await
关键字则用于暂停当前协程的执行，等待另一个协程或可等待对象（如I/O操作）完成。当一个协程遇到
await
时，它会交出控制权给事件循环，让事件循环去执行其他准备好的协程。一旦
await
等待的对象准备就绪，事件循环就会把控制权还给这个协程，让它从上次暂停的地方继续执行。
asyncio则是这一切的幕后英雄。它提供了一个事件循环（Event loop），这个循环会不断地检查哪些任务可以运行，哪些任务正在等待I/O。它负责注册、调度和执行协程。当你用
asyncio.run()
启动一个主协程时，实际上就是启动了事件循环。这个循环会一直运行，直到所有注册的任务都完成。它管理着网络连接、文件I/O等异步操作，确保它们在不阻塞主线程的情况下进行。这种“协作式多任务”模型，让python程序在处理大量并发I/O请求时，能够表现出极高的效率和响应速度。我觉得，这就像一个高明的管家，在等待客人（I/O）的时候，绝不会闲着，而是会去处理其他事情，效率自然就上来了。
Python协程与多线程/多进程有何本质差异？
这真的是一个老生常谈，却又常常让人混淆的问题。在我看来，Python协程与多线程或多进程最根本的区别在于它们的并发模型和资源消耗。
立即学习“Python免费学习笔记（深入）”；
多线程和多进程是操作系统层面的并发，它们是“抢占式”的。操作系统负责调度，随时可以中断一个线程或进程的执行，去运行另一个。多进程有独立的内存空间，隔离性好，但创建和切换开销大；多线程共享内存，开销相对小，但会面临全局解释器锁（GIL）的限制，导致在CPU密集型任务上无法真正并行利用多核。更头疼的是，共享数据带来的同步问题，锁、信号量、死锁，这些都是多线程编程的噩梦。
而协程，它是“协作式”的并发，运行在单个线程内。它完全由程序自身控制调度，一个协程只有在遇到
await
时才会主动让出控制权。这意味着，只要一个协程不主动让出，它就会一直执行下去，不会被“抢占”。因此，协程天然没有多线程那样的GIL限制（因为只有一个线程），也不存在复杂的共享数据同步问题（因为没有真正的并行）。它的创建和切换开销极小，因为它只是函数栈帧的切换，而不是操作系统线程或进程的上下文切换。
所以，如果你面对的是大量I/O等待（比如网络请求、文件读写、数据库查询），协程是绝佳选择，它能高效地利用CPU在等待期间处理其他任务。但如果你的任务是CPU密集型的，需要大量计算，那么协程就帮不上什么忙了，因为它仍然运行在单个CPU核心上，此时多进程才是王道，可以真正利用多核进行并行计算。我有时会想，协程就像一个高情商的团队成员，懂得在自己忙不过来时，主动把机会让给别人，而多线程更像是一群有点“争抢”的团队成员，需要一个严格的领导（操作系统）来协调。
在实际项目中，asyncio能解决哪些具体的开发痛点？
asyncio在实际项目中的应用场景非常广泛，尤其是在需要处理大量并发I/O的场景下，它的优势能得到充分体现。
一个非常典型的痛点是Web爬虫。想象一下，你需要从成千上万个网页上抓取数据。如果使用同步请求，一个接一个地访问，效率会非常低下，因为大部分时间都花在等待网络响应上了。而使用asyncio，你可以同时发起数百甚至数千个http请求，当一个请求在等待响应时，事件循环会去处理其他请求，大大缩短了总体的爬取时间。我之前做过一个数据采集项目，从同步切换到asyncio后，效率提升了不止一个数量级，那种感觉就像从龟速拨号上网突然升级到了光纤。
另一个痛点是构建高性能网络服务。无论是API服务器、websocket服务器还是其他自定义协议的服务器，都需要能够同时处理大量客户端连接。传统的同步服务器模型，一个连接往往会占用一个线程或进程，资源消耗大，并发能力有限。asyncio提供了一套非阻塞的网络I/O接口，可以轻松构建单线程但高并发的网络服务。例如，我们可以用
aiohttp
构建异步Web服务，用
websockets
库构建异步WebSocket应用，它们都能以极低的资源消耗承载海量的并发连接。
此外，数据库操作也是asyncio大显身手的地方。很多现代数据库驱动都提供了异步接口（如
asyncpg
aiomysql
for MySQL），这意味着你可以在等待数据库查询结果时，不阻塞整个应用程序，从而提高数据库密集型应用的响应速度。还有，长耗时的后台任务，比如批量数据处理、消息队列消费者等，如果它们内部包含I/O操作，用asyncio来编写，可以确保它们在后台高效运行，而不会影响到用户界面的响应或主服务的性能。
如何在asyncio应用中有效管理并发任务与异常？
在asyncio的世界里，管理并发任务和处理异常是构建健壮应用的关键。毕竟，我们不是在写简单的脚本，而是要处理各种复杂情况。
对于并发任务的管理，asyncio提供了几个核心工具。
asyncio.create_task()
是最基础的，它用于将一个协程函数包装成一个任务，并提交给事件循环运行。当你需要同时运行多个独立的协程，并且不需要等待它们全部完成时，这是一个很好的选择。但如果我们需要等待所有任务都完成，并且可能需要收集它们的结果，那么
asyncio.gather()
就派上用场了。
import asyncio async def fetch_data(delay, id): print(f"Task {id}: 开始获取数据，预计延迟 {delay}s") await asyncio.sleep(delay) print(f"Task {id}: 数据获取完成") return f"Data from {id} after {delay}s" async def main_gather(): tasks = [ fetch_data(2, "A"), fetch_data(1, "B"), fetch_data(3, "C") ] # 等待所有任务完成，并收集结果 results = await asyncio.gather(*tasks) print(f"所有任务完成，结果：{results}") # asyncio.run(main_gather())
asyncio.as_completed()
则提供了一种不同的并发模式，它返回一个迭代器，每次迭代都会按完成顺序返回一个已完成任务的Future对象。这在你不需要等待所有任务，而是想尽快处理已完成任务的结果时非常有用。
至于异常处理，这在异步编程中尤为重要。一个未捕获的异常可能会导致整个事件循环崩溃。最直接的方式是在
await
调用周围使用标准的
try...except
块，就像处理同步代码一样。
async def might_fail_task(id): print(f"Task {id}: 尝试执行可能失败的任务") if id == "B": raise ValueError(f"Task {id} 故意抛出错误") await asyncio.sleep(1) return f"Task {id} 成功" async def main_exception_individual(): try: result_a = await might_fail_task("A") print(result_a) except ValueError as e: print(f"捕获到异常: {e}") try: result_b = await might_fail_task("B") # 这个会失败 print(result_b) except ValueError as e: print(f"捕获到异常: {e}") # asyncio.run(main_exception_individual())
当使用
asyncio.gather()
时，异常处理会稍微复杂一些。默认情况下，如果
gather
中的任何一个任务抛出异常，那么
gather
本身也会立即抛出第一个遇到的异常，而其他未完成的任务则会被取消。但你可以通过设置
return_exceptions=True
来改变这种行为。在这种模式下，即使有任务抛出异常，
gather
也会等待所有任务完成，并将异常作为结果列表中的一项返回，而不是直接抛出。这对于批量处理任务，即使部分失败也希望获取所有结果的场景非常有用。
async def main_exception_gather(): tasks = [ might_fail_task("X"), might_fail_task("Y"), might_fail_task("Z") ] # 默认行为，第一个异常会直接抛出 # try: # results = await asyncio.gather(*tasks) # print(f"所有任务完成，结果：{results}") # except ValueError as e: # print(f"捕获到gather中的异常 (默认行为): {e}") # 使用return_exceptions=True，异常作为结果返回 results_with_exceptions = await asyncio.gather(*tasks, return_exceptions=True) print(f"所有任务完成 (带异常返回)，结果：{results_with_exceptions}") for res in results_with_exceptions: if isinstance(res, Exception): print(f"发现一个任务失败: {res}") else: print(f"一个任务成功: {res}") # asyncio.run(main_exception_gather())
在我看来，掌握这些并发和异常处理的技巧，是真正发挥asyncio威力的关键。它能让你在构建高并发应用时，既能享受其带来的性能优势，又能确保代码的健壮性和可维护性。
评论（已关闭）
评论已关闭