Python 异步编程实战

异步编程是现代 Python 开发中不可或缺的技能。随着 FastAPI 等异步框架的流行，理解 asyncio 的工作原理变得越来越重要。本文将从基础概念出发，逐步深入到实际项目中的应用。

为什么需要异步编程

传统的同步编程模型在处理 I/O 密集型任务时会浪费大量时间等待。比如一个 Web 服务需要查询数据库，在等待数据库返回结果的几十毫秒里， CPU 完全空闲。异步编程的核心思想就是在等待期间让 CPU 去处理其他任务，从而大幅提升吞吐量。

下面的同步代码依次发起三个 HTTP 请求，每个请求耗时 1 秒，总计需要 3 秒：

import time
import requests

def fetch_sync(url):
    return requests.get(url).text

def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
    ]
    start = time.time()
    for url in urls:
        fetch_sync(url)
    print(f"耗时: {time.time() - start:.2f}s")  # 约 3 秒

main()

asyncio 基础

asyncio 核心概念：

协程（Coroutine）：用 async def 定义的函数，可以在执行中暂停和恢复
事件循环（Event Loop）：调度和执行协程的核心引擎
任务（Task）：对协程的封装，用于并发调度

import asyncio
import time
import aiohttp

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

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/1",
    ]
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_async(session, url) for url in urls]
        await asyncio.gather(*tasks)
    print(f"耗时: {time.time() - start:.2f}s")  # 约 1 秒

asyncio.run(main())

三个请求并发执行，总耗时仅约 1 秒。

创建和管理任务

async def worker(name, delay):
    await asyncio.sleep(delay)
    print(f"{name} 完成")
    return name

async def main():
    task1 = asyncio.create_task(worker("任务A", 2))
    task2 = asyncio.create_task(worker("任务B", 1))

    print("等待任务完成...")
    results = await asyncio.gather(task1, task2)
    print(f"结果: {results}")

asyncio.run(main())

同步原语

# Semaphore 限制并发数
sem = asyncio.Semaphore(5)

async def fetch_with_limit(session, url):
    async with sem:
        async with session.get(url) as resp:
            return await resp.text()

实战：FastAPI 构建异步 Web 服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncpg
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(
        "postgresql://user:pass@localhost/db",
        min_size=5, max_size=20
    )
    yield
    await app.state.pool.close()

app = FastAPI(lifespan=lifespan)

class UserOut(BaseModel):
    id: int
    name: str
    email: str

@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
    async with app.state.pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, name, email FROM users WHERE id = $1", user_id
        )
    if row is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    return dict(row)

常见陷阱与最佳实践

不要在协程中调用阻塞函数。使用 asyncio.to_thread() 将阻塞操作放到线程池。
始终使用连接池。无论数据库还是 HTTP 客户端，复用连接减少开销。
合理设置并发上限。使用 Semaphore 控制并发数量。
正确处理取消。使用 try/finally 确保资源在任务取消时得到清理。

总结

Python 的异步生态已经非常成熟。从底层的 asyncio 到上层的 FastAPI，开发者可以根据需求选择合适的抽象层次。核心要点：理解事件循环工作机制，避免阻塞调用，善用并发控制原语。

← 返回首页