Python 异步编程：async/await 完全指南

为什么需要异步？

在 I/O 密集型场景中，传统同步代码会浪费大量时间等待：

# 同步方式 - 串行执行
import time

def fetch_data():
    time.sleep(2)  # 模拟网络请求
    return "data"

start = time.time()
result1 = fetch_data()
result2 = fetch_data()
result3 = fetch_data()
print(f"耗时：{time.time() - start:.2f}秒")  # 约 6 秒

async/await 基础

import asyncio

async def fetch_data():
    await asyncio.sleep(2)  # 非阻塞等待
    return "data"

async def main():
    # 并发执行
    results = await asyncio.gather(
        fetch_data(),
        fetch_data(),
        fetch_data()
    )
    print(results)  # ["data", "data", "data"]

asyncio.run(main())  # 耗时约 2 秒！

核心概念

1. Coroutine（协程）

• 使用 async def 定义
• 调用时不会立即执行，返回 coroutine 对象
• 需要用 await 或 asyncio.create_task() 启动

2. Event Loop（事件循环）

• 调度和运行协程的核心
• asyncio.run() 会自动创建和关闭

3. Task（任务）

• 包装协程，支持并发执行
• 可以取消、获取状态

实战场景

并发 API 调用

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

urls = ["http://api1.com", "http://api2.com", "http://api3.com"]
results = asyncio.run(fetch_all(urls))

异步数据库操作

import asyncpg

async def query_users():
    conn = await asyncpg.connect(
        host="localhost",
        user="postgres",
        password="secret",
        database="mydb"
    )
    rows = await conn.fetch("SELECT * FROM users")
    await conn.close()
    return rows

常见陷阱

❌ 忘记 await

async def wrong():
    result = fetch_data()  # 返回 coroutine，不会执行！

❌ 阻塞操作污染异步代码

async def wrong():
    time.sleep(1)  # 阻塞整个事件循环！
    await asyncio.sleep(1)  # 正确做法

✅ 正确做法

async def correct():
    result = await fetch_data()
    await asyncio.sleep(1)

性能对比

场景	同步	异步
100 个 API 调用	~200 秒	~2 秒
数据库批量查询	串行等待	并发执行
文件批量处理	阻塞 I/O	非阻塞 I/O

最佳实践

1. I/O 密集型用 async：网络、数据库、文件
2. CPU 密集型用多进程：async 不提升 CPU 性能
3. 混合使用：用 run_in_executor 处理阻塞代码
4. 超时控制：用 asyncio.wait_for() 设置超时

try:
    result = await asyncio.wait_for(fetch_data(), timeout=5.0)
except asyncio.TimeoutError:
    print("请求超时")

结语

异步编程是 Python 高性能的关键技能。掌握 async/await，让你的程序效率提升 10 倍！

---

有问题欢迎在评论区交流！