Python 异步编程：asyncio 与 FastAPI 实战指南

1. 异步编程核心概念

• 事件循环：异步任务调度核心，通过$ \text{event_loop} = \text{asyncio.get_event_loop()} $ 管理任务队列
• 协程（Coroutine）：使用 async def 定义，通过 await 挂起阻塞操作
• 任务（Task）：协程的封装，$ \text{task} = \text{asyncio.create_task(coroutine())} $

2. asyncio 基础实战

import asyncio

async def fetch_data(delay: int, id: int):
    print(f"任务 {id} 启动")
    await asyncio.sleep(delay)  # 模拟IO操作
    return f"数据{id}"

async def main():
    # 并发执行任务
    tasks = [fetch_data(2-i, i) for i in range(3)]
    results = await asyncio.gather(*tasks)
    print(f"结果集: {results}")

asyncio.run(main())

输出：

任务 0 启动
任务 1 启动
任务 2 启动
结果集: ['数据0', '数据1', '数据2']

3. FastAPI 异步集成

FastAPI 原生支持异步路由：

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/async-data")
async def get_data():
    await asyncio.sleep(1)  # 模拟数据库查询
    return {"status": "success", "data": [1, 2, 3]}

@app.get("/sync-data")
def sync_data():  # 同步路由对比
    time.sleep(1)
    return {"status": "success"}

4. 性能优化关键点

1. IO密集型场景：数据库查询、API调用等阻塞操作使用 await $$ \text{吞吐量} \propto \frac{1}{\text{阻塞时间}} $$
2. 并发控制：使用信号量限制并发数

   semaphore = asyncio.Semaphore(10)
   async with semaphore:
       await db_query()
   

3. 任务取消：通过 $ \text{task.cancel()} $ 实现超时中断

5. 完整实战案例：异步天气API

from fastapi import FastAPI
import httpx

app = FastAPI()

async def fetch_weather(city: str):
    async with httpx.AsyncClient() as client:
        url = f"https://api.weather.com/{city}"
        resp = await client.get(url, timeout=3.0)
        return resp.json()

@app.get("/weather/{city}")
async def get_weather(city: str):
    data = await fetch_weather(city)
    return {"city": city, "temp": data["temp"]}

优势：

• 单线程处理数百并发请求
• 响应时间降低至同步模式的$ \frac{1}{N} $（N为并发数）

调试提示：使用 uvicorn main:app --reload 启动时添加 --log-level debug 查看异步任务调度日志