Python Asyncio 实战指南：从入门到放弃再到真香引言大家好，我是默然，今天给大家分享Python异步编程的实战经验。很多Python开发者对asyncio都又爱又恨，觉得它复杂难用，坑很多，学习曲线陡峭。我当初也是从入门到放弃，后来在项目中被逼着用了几次，才慢慢体会到异步编程的好处，现在已经是真香了。今天我就用最通俗的语言，带大家搞懂asyncio，看完就能在项目中用起来。正文部分1. 什么是异步编程？我们为什么需要它？在讲asyncio之前，我们先搞清楚什么是异步编程，它能解决什么问题。我们平时写的Python代码大部分都是同步的，也就是代码一行一行顺序执行，上一行没执行完，下一行就等着。比如你要调用一个接口，网络请求需要1秒，这时候整个程序就卡住1秒，什么都干不了，只能等请求返回。如果你的程序大部分时间都在等待IO（比如网络请求、数据库查询、文件读写），那同步模式就会浪费大量的CPU时间，这时候异步编程就派上用场了。异步编程的核心思想就是：在等待IO操作的时候，程序不闲着，去干点别的事情，等IO操作完成了再回来继续处理。举个通俗的例子：同步模式：你烧水的时候站在水壶旁边等着，水开了再去泡茶，泡茶的时候等着茶泡好，再去吃早餐，整个过程要30分钟异步模式：你先把水烧上，然后去刷牙，刷完牙水刚好开了，泡上茶，然后去吃早餐，早餐吃完茶也泡好了，整个过程只需要15分钟同样的事情，异步模式效率提升了一倍！这就是异步编程的优势。2. Asyncio 核心概念快速搞懂asyncio里有几个核心概念，搞懂了这几个概念，你就理解了80%的asyncio。2.1 协程（Coroutine）协程是asyncio的基本单位，你可以把它理解成一个"可以暂停的函数"。用async def定义的函数就是协程函数：async def hello(): print("Hello") await asyncio.sleep(1) print("World")注意：直接调用协程函数不会执行，它只会返回一个协程对象，需要放到事件循环里才能执行。2.2 事件循环（Event Loop）事件循环是asyncio的核心，你可以把它理解成一个死循环，不断地检查有没有要执行的任务，有就执行，没有就等着。import asyncio # 获取事件循环 loop = asyncio.get_event_loop() # 运行协程 loop.run_until_complete(hello()) # 关闭循环 loop.close()Python 3.7+之后可以用更简单的asyncio.run()：asyncio.run(hello())2.3 Future 和 TaskFuture：代表一个异步操作的结果，还没完成的时候是pending状态，完成了是done状态Task：是Future的子类，用来包装协程，把协程放到事件循环里调度执行我们可以用asyncio.create_task()来创建任务，让多个协程并发执行：import asyncio async def task1(): await asyncio.sleep(2) print("任务1完成") async def task2(): await asyncio.sleep(1) print("任务2完成") async def main(): # 创建两个任务，并发执行 t1 = asyncio.create_task(task1()) t2 = asyncio.create_task(task2()) # 等待两个任务都完成 await t1 await t2 asyncio.run(main())运行结果：任务2完成 任务1完成两个任务总共只花了2秒，而不是3秒，这就是并发的效果！2.4 await 关键字await是用来暂停协程执行的，当你await一个协程、Task或者Future的时候，当前协程会暂停，把CPU让给其他协程，等await的对象完成了再回来继续执行。注意：await只能在async函数里面用，在普通函数里用会报错。3. 常见使用场景和示例asyncio特别适合IO密集型的任务，下面是几个常见的使用场景。3.1 并发爬取网页这是asyncio最常用的场景之一，比如你要爬取100个网页，用同步的方式要等100个请求依次完成，用异步的方式可以同时发送多个请求。我们用aiohttp这个异步HTTP库来演示：import asyncio import aiohttp async def fetch_url(session, url): try: async with session.get(url, timeout=10) as response: content = await response.text() print(f"成功爬取 {url}, 长度: {len(content)}") return content except Exception as e: print(f"爬取 {url} 失败: {e}") return None async def main(): urls = [ "https://www.baidu.com", "https://www.zhihu.com", "https://www.github.com", "https://www.python.org", "https://www.moranblog.cn" ] async with aiohttp.ClientSession() as session: # 创建任务列表 tasks = [fetch_url(session, url) for url in urls] # 并发执行所有任务 results = await asyncio.gather(*tasks) print(f"\n所有任务完成，共爬取 {len([r for r in results if r])} 个页面") if __name__ == "__main__": asyncio.run(main())100个网页如果用同步方式可能需要几十秒，用异步方式几秒钟就搞定了。3.2 异步操作数据库很多数据库都有异步驱动，比如asyncpg（PostgreSQL）、aiomysql（MySQL）、aioredis（Redis），用异步驱动可以大大提升数据库操作的吞吐量。比如用aioredis操作Redis：import asyncio import aioredis async def main(): # 连接Redis redis = aioredis.from_url("redis://localhost") # 异步写数据 await redis.set("name", "默然") await redis.set("age", "18") # 异步读数据 name = await redis.get("name", encoding="utf-8") age = await redis.get("age", encoding="utf-8") print(f"name: {name}, age: {age}") # 关闭连接 await redis.close() asyncio.run(main())3.3 异步API服务用FastAPI这个异步框架写API服务，性能可以和Go、Node.js媲美，非常适合写高并发的后端接口。一个简单的FastAPI示例：from fastapi import FastAPI import asyncio app = FastAPI() @app.get("/") async def root(): # 模拟一个耗时的IO操作 await asyncio.sleep(1) return {"message": "Hello World"} @app.get("/items/{item_id}") async def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q}用uvicorn启动：uvicorn main:app --reloadFastAPI会自动处理异步请求，支持高并发。4. Asyncio 常见的坑和避坑指南asyncio虽然强大，但也有很多坑，我总结了几个最常见的，大家一定要注意。4.1 不要在异步代码里调用同步阻塞函数这是新手最容易犯的错误！如果你在async函数里调用了一个同步的阻塞函数，比如time.sleep(1)、同步的requests请求、同步的数据库查询，那整个事件循环都会被卡住，异步就白用了。❌ 错误的写法：import asyncio import time async def bad_example(): print("开始") # 这里会阻塞整个事件循环！ time.sleep(1) print("结束")✅ 正确的写法：async def good_example(): print("开始") # 用异步的sleep await asyncio.sleep(1) print("结束")如果必须要用同步函数怎么办？可以用asyncio.to_thread()把它放到线程里执行：async def good_example(): print("开始") # 把同步函数放到线程里执行，不会阻塞事件循环 await asyncio.to_thread(time.sleep, 1) print("结束")4.2 不要用太多的Task，控制并发数很多人以为并发数越多越快，其实不是这样的，并发数太高会导致频繁的上下文切换，反而会降低性能，还有可能把目标服务器打垮。比如你要爬取1000个网页，不要一下子创建1000个Task，最好用信号量控制并发数：import asyncio import aiohttp # 控制最多同时10个并发 semaphore = asyncio.Semaphore(10) async def fetch_url(session, url): async with semaphore: try: async with session.get(url, timeout=10) as response: return await response.text() except Exception as e: print(f"爬取失败: {e}") return None这样就不会一下子发出去1000个请求了。4.3 异常处理要做好异步代码的异常处理和同步代码有点不一样，如果你不处理Task里的异常，程序崩溃了可能都不知道为什么。有几种处理异常的方式：在协程内部用try/except捕获用asyncio.gather(return_exceptions=True)把异常作为结果返回给Task添加异常回调示例：async def may_fail(): raise RuntimeError("出错了") async def main(): # 方式1：捕获gather的异常 try: results = await asyncio.gather(may_fail(), may_fail()) except Exception as e: print(f"捕获到异常: {e}") # 方式2：把异常作为结果返回 results = await asyncio.gather(may_fail(), may_fail(), return_exceptions=True) for result in results: if isinstance(result, Exception): print(f"处理异常: {result}")4.4 注意asyncio的版本差异asyncio从Python 3.4到现在变化很大，很多API都变了：Python 3.4：引入asyncio，用@asyncio.coroutine和yield fromPython 3.5：引入async/await语法Python 3.7：加入asyncio.run()，简化启动Python 3.10：加入asyncio.timeout()，很多API优化Python 3.11：性能大幅提升，比3.10快30%左右建议尽量用Python 3.10+的版本，很多新特性和性能优化非常香。4.5 调试异步代码很头疼异步代码的调试确实比同步代码难，调用栈很乱，不过还是有一些方法：用asyncio.debug()开启调试模式，可以看到慢任务、未处理的异常等用print打日志比断点调试好用用py-spy等性能分析工具看哪里慢5. 什么时候不该用asyncio？asyncio不是银弹，不是所有场景都适合用：CPU密集型任务：比如大量的计算、图像处理，这时候用多进程更合适，asyncio不适合简单的小脚本：只有几个逻辑，用同步方式更简单，没必要搞异步第三方库不支持异步：如果你的依赖库大部分都是同步的，强行用异步会非常痛苦，到处都是to_thread，反而降低代码可读性6. 学习资源推荐官方文档：https://docs.python.org/zh-cn/3/library/asyncio.html （写得非常好，建议多看）Asyncio 官方教程：https://docs.python.org/zh-cn/3/library/asyncio-task.htmlaiohttp 文档：https://docs.aiohttp.org/FastAPI 文档：https://fastapi.tiangolo.com/zh/总结Asyncio确实有一定的学习曲线，一开始会觉得很绕，很多概念搞不懂，但一旦你掌握了它，就能写出高性能的IO密集型应用，性能提升非常明显。不要因为一开始觉得难就放弃，多写几个小项目练手，很快就能熟练掌握。现在Python的异步生态已经非常成熟了，无论是爬虫、API服务、消息队列、物联网等场景都有很好的支持，非常值得学习。如果在使用asyncio的过程中遇到什么问题，欢迎在评论区留言交流。如果你觉得我的文章对你有用，请随意赞赏，也欢迎分享给更多需要的朋友。最后修改：2026 年 03 月 13 日© 允许规范转载