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 和 Task

• Future：代表一个异步操作的结果，还没完成的时候是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 --reload

FastAPI会自动处理异步请求，支持高并发。

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里的异常，程序崩溃了可能都不知道为什么。

有几种处理异常的方式：

1. 在协程内部用try/except捕获
2. 用asyncio.gather(return_exceptions=True)把异常作为结果返回
3. 给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 from
• Python 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 调试异步代码很头疼

异步代码的调试确实比同步代码难，调用栈很乱，不过还是有一些方法：

1. 用asyncio.debug()开启调试模式，可以看到慢任务、未处理的异常等
2. 用print打日志比断点调试好用
3. 用py-spy等性能分析工具看哪里慢

5. 什么时候不该用asyncio？

asyncio不是银弹，不是所有场景都适合用：

1. CPU密集型任务：比如大量的计算、图像处理，这时候用多进程更合适，asyncio不适合
2. 简单的小脚本：只有几个逻辑，用同步方式更简单，没必要搞异步
3. 第三方库不支持异步：如果你的依赖库大部分都是同步的，强行用异步会非常痛苦，到处都是to_thread，反而降低代码可读性

6. 学习资源推荐

• 官方文档：https://docs.python.org/zh-cn/3/library/asyncio.html （写得非常好，建议多看）
• Asyncio 官方教程：https://docs.python.org/zh-cn/3/library/asyncio-task.html
• aiohttp 文档：https://docs.aiohttp.org/
• FastAPI 文档：https://fastapi.tiangolo.com/zh/

总结

Asyncio确实有一定的学习曲线，一开始会觉得很绕，很多概念搞不懂，但一旦你掌握了它，就能写出高性能的IO密集型应用，性能提升非常明显。

不要因为一开始觉得难就放弃，多写几个小项目练手，很快就能熟练掌握。现在Python的异步生态已经非常成熟了，无论是爬虫、API服务、消息队列、物联网等场景都有很好的支持，非常值得学习。

如果在使用asyncio的过程中遇到什么问题，欢迎在评论区留言交流。

如果你觉得我的文章对你有用，请随意赞赏，也欢迎分享给更多需要的朋友。

最后修改：2026 年 03 月 13 日
© 允许规范转载