一、引言：为什么需要异步编程

在现代 Web 开发中，高并发 IO 密集型任务是最常见的场景——处理数千个 HTTP 请求、读写数据库、调用外部 API。传统的同步阻塞模型在面对这类任务时，线程上下文切换的开销会急剧上升，导致资源利用率低。

Python 的 asyncio 模块提供了一种基于事件循环的并发模型，通过单线程内的协作式多任务处理，以极低的开销实现高并发。与多线程不同，asyncio 没有线程切换和锁竞争的开销，是 IO 密集型任务的最佳选择。

二、事件循环（Event Loop）核心原理

事件循环是 asyncio 的心脏。它本质上是一个死循环，持续从任务队列中取出就绪的协程并执行，同时监听 IO 事件。

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

# Python 3.7+
asyncio.run(hello())

执行流程如下：

1. asyncio.run() 创建事件循环
2. hello() 协程被调度执行，打印 "Hello"
3. 遇到 await asyncio.sleep(1)，协程挂起，控制权交还给事件循环
4. 1 秒后，事件循环唤醒协程，继续执行打印 "World"
5. 协程结束，事件循环关闭

2.1 事件循环的内部机制

底层实现中，asyncio 的事件循环基于 selectors 模块或 IOCP（Windows）实现 IO 多路复用。它维护着三个核心队列：

队列	用途
ready	就绪队列，存放可以立即执行的回调
scheduled	定时队列，存放等待超时的回调
io_events	IO 事件队列，存放等待文件描述符事件的回调

每次循环迭代时，事件循环：

1. 从 scheduled 中取出已超时的回调，移入 ready
2. 从 io_events 中取出已就绪的 IO 回调，移入 ready
3. 执行 ready 中的所有回调
4. 挂起等待新的事件（如果没有就绪任务）

三、协程详解

Python 3.5 引入了 async def 和 await 关键字，正式支持协程。协程是一种可以暂停和恢复的函数，其状态保存在帧对象（frame object）中。

3.1 协程 vs 生成器

特性	生成器	协程
定义	yield	async def
返回值	Iterator	Coroutine
暂停机制	yield	await
用途	惰性求值	异步 IO
驱动方式	手动 next()	事件循环

3.2 协程的生命周期

async def fetch_data(url):
    print(f"开始请求: {url}")
    data = await http_get(url)  # 挂起点
    print(f"请求完成: {url}")
    return data

coro = fetch_data("/api/user")  # 创建协程对象，未执行
# 协程处于 CORO_CREATED 状态

asyncio.run(coro)  # 调度执行
# 执行中：CORO_RUNNING
# 挂起中：CORO_SUSPENDED
# 完成：CORO_CLOSED

四、Task 与 Future

Task 是 Future 的子类，它将协程包装为可调度的任务，提供结果获取、取消和回调注册功能。

4.1 创建任务

import asyncio

async def worker(name, delay):
    await asyncio.sleep(delay)
    return f"Worker {name} done"

async def main():
    # 同时创建三个任务并发执行
    tasks = [
        asyncio.create_task(worker("A", 3)),
        asyncio.create_task(worker("B", 1)),
        asyncio.create_task(worker("C", 2)),
    ]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"完成: {result}")

asyncio.run(main())
# 输出顺序: B, C, A

4.2 Task 的取消与超时

async def timeout_demo():
    task = asyncio.create_task(slow_operation())
    try:
        result = await asyncio.wait_for(task, timeout=5.0)
    except asyncio.TimeoutError:
        print("操作超时，已取消")
        # task 会被自动取消
    # 也可以显式取消
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("任务已被取消")

五、实战：构建高性能异步 Web 服务

使用 FastAPI 和 httpx 构建一个异步 Web 服务：

from fastapi import FastAPI, HTTPException
import httpx
import asyncio

app = FastAPI()
client = httpx.AsyncClient()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 并发请求多个下游服务
    async def fetch_orders():
        resp = await client.get(f"http://orders/api/users/{user_id}/orders")
        return resp.json()

    async def fetch_profile():
        resp = await client.get(f"http://profile/api/users/{user_id}")
        return resp.json()

    orders, profile = await asyncio.gather(
        fetch_orders(),
        fetch_profile(),
        return_exceptions=True
    )

    if isinstance(orders, Exception):
        orders = []
    if isinstance(profile, Exception):
        raise HTTPException(status_code=502, detail="Profile service unavailable")

    return {"user": profile, "orders": orders}

5.1 信号量限流

防止突发请求压垮下游服务：

semaphore = asyncio.Semaphore(10)

async def rate_limited_fetch(url):
    async with semaphore:
        return await client.get(url)

async def batch_fetch(urls):
    tasks = [rate_limited_fetch(url) for url in urls]
    return await asyncio.gather(*tasks)

六、异步数据库操作

使用 asyncpg 和 SQLAlchemy 1.4+ 的异步支持：

import asyncpg
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

# asyncpg 原生方式
async def query_with_asyncpg():
    conn = await asyncpg.connect(
        user="user", password="pass",
        database="db", host="localhost"
    )
    rows = await conn.fetch("SELECT * FROM users WHERE active = $1", True)
    await conn.close()
    return rows

# SQLAlchemy 异步方式
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession)

async def query_with_sa():
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            text("SELECT * FROM users WHERE active = :active"),
            {"active": True}
        )
        return result.all()

七、常见陷阱与最佳实践

7.1 不要在协程中使用阻塞调用

阻塞调用会阻塞整个事件循环，导致所有并发任务停滞：

# ❌ 错误：阻塞调用会阻塞事件循环
async def bad():
    time.sleep(1)  # 阻塞！事件循环停转

# ✅ 正确：使用异步 sleep
async def good():
    await asyncio.sleep(1)

7.2 使用 asyncio.to_thread 封装阻塞 IO

# 对于无法替换为异步库的阻塞调用
async def handle_request():
    # 在独立线程中运行阻塞函数，不阻塞事件循环
    result = await asyncio.to_thread(blocking_io_func, arg1, arg2)
    return result

7.3 资源清理

始终正确清理资源：

# 使用 async with 确保资源释放
async def safe_operation():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com") as resp:
            return await resp.json()
    # 离开 async with 块后自动释放连接

八、性能调优

优化项	说明	效果
连接池复用	复用 HTTP/TCP 连接，避免三次握手	延迟降低 30-50%
限制并发数	使用 Semaphore 防止资源耗尽	稳定性提升
批量处理	将小任务合并为批次	吞吐量提升 2-5x
使用 uvloop	替代内置事件循环	性能提升 2-4x
取消不必要任务	及时取消超时/无用任务	内存占用降低

import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# uvloop 基于 libuv，性能接近 Go 的 goroutine

九、总结

Python asyncio 提供了一套完整的异步编程生态。掌握其核心概念——事件循环、协程、Task、Future——是写出高效异步代码的基础。结合实际项目中的限流、超时控制、资源管理等最佳实践，可以充分发挥异步编程在高并发场景下的优势。

值得注意的是，asyncio 虽然强大，但并不适用于 CPU 密集型任务。对于计算密集型场景，多进程（multiprocessing）或分布式计算仍然是更合适的选择。

随着 Python 生态的发展，越来越多主流库提供了异步支持（FastAPI、SQLAlchemy、httpx、asyncpg 等），异步编程已成为现代 Python 开发者的必备技能。