FastAPI异步编程完全指南

📂 所属阶段：第二阶段 — 进阶黑科技（核心篇）
🔗 相关章节：FastAPI依赖注入系统 · FastAPI中间件应用

目录

什么是异步编程？

同步vs异步：排队打饭的比喻 🍚

想象你去食堂打饭：

  • 同步：你站在窗口等厨师炒完一盘菜，端走，再点下一道。厨师在炒菜时你只能干等。
  • 异步：你把菜单交给厨师，然后去拿餐具、倒饮料，等菜好了服务员直接端到你桌上。你不需要傻站着等。

这就是核心区别：等待时是否可以切换去干别的事。

为什么Web服务需要异步？

请求 A（查询数据库，耗时 200ms）
请求 B（简单计算，耗时 1ms）
请求 C（调用外部 API，耗时 1000ms）
  • 同步模式：按顺序处理，总耗时 = 200 + 1 + 1000 = 1201ms
  • 异步模式：A等待数据库时切换B，B完成切C，C等待时切回A → 总耗时≈1000ms

在I/O密集型场景（数据库查询、HTTP请求、文件读写），异步能让单线程处理大量并发请求，显著提升API性能。

异步编程的优势

  1. 高并发处理能力：单个进程可以处理数千个并发连接
  2. 资源利用率高：避免线程创建和切换的开销
  3. 响应速度快：I/O等待期间可以处理其他请求
  4. 内存占用低：协程比线程占用更少内存

async/await详解

基础语法

import asyncio

# 定义一个协程函数（async def）
async def say_hello():
    print("Hello!")
    return "Hello from coroutine!"

# 运行协程的三种方式
# 方式一：asyncio.run()（推荐，主入口用）
asyncio.run(say_hello())

# 方式二：在已有事件循环中创建任务
async def main():
    task = asyncio.create_task(say_hello())  # 调度执行协程
    result = await task                       # 等待任务完成
    print(f"Result: {result}")

asyncio.run(main())

await 在等什么？

await 只能等待可等待对象（Awaitable），包括：

类型示例说明
协程（Coroutine）await coro()async def 返回的对象
任务（Task）asyncio.create_task()调度好的协程
Futureasyncio.Future()尚未完成的结果占位符

asyncio.sleep vs time.sleep

import asyncio
import time

# ❌ 同步sleep：阻塞整个线程
def sync_task():
    time.sleep(2)  # 2秒内整条线程被卡住
    print("Sync done")

# ✅ 异步sleep：让出控制权，处理其他协程
async def async_task():
    await asyncio.sleep(2)  # 2秒内线程可以干别的
    print("Async done")

# 体验差异
async def compare():
    start = time.time()
    await asyncio.gather(async_task(), async_task())  # 并发执行
    print(f"并发耗时: {time.time() - start:.2f}s")  # ~2秒

asyncio.run(compare())

⚠️ 在FastAPI中用time.sleep()会阻塞整个事件循环！

事件循环原理

什么是事件循环？

事件循环是异步的"调度中心"，工作流程：

┌─────────────────────────────────────┐
│          事件循环（单线程）             │
├─────────────────────────────────────┤
│  1. 检查IO事件（网络/文件/定时器）        │
│  2. 收集就绪任务，执行它们               │
│  3. 遇到await就挂起，切换到下一个任务     │
│  4. 重复...直到全部完成                  │
└─────────────────────────────────────┘

事件循环生命周期

import asyncio

async def task(name, sec):
    print(f"[{name}] 开始")
    await asyncio.sleep(sec)
    print(f"[{name}] 完成")

async def main():
    t1 = asyncio.create_task(task("A", 1))
    t2 = asyncio.create_task(task("B", 0.5))
    t3 = asyncio.create_task(task("C", 0.8))
    await asyncio.gather(t1, t2, t3)
    print("全部完成")

asyncio.run(main())
# 输出顺序：[B]开始→[A]开始→[C]开始→[B]完成→[C]完成→[A]完成

异步函数调用规则

四条黄金法则

调用方被调函数正确写法
sync函数async函数❌ 不能直接调用
sync函数sync函数✅ 直接调用
async函数async函数✅ await调用
async函数sync函数⚠️ 可以，但耗时同步操作需丢线程池

异步中调用耗时同步函数

import asyncio
import time

def blocking_function():
    time.sleep(1)
    return "Blocking result"

# ✅ 推荐：Python 3.9+用to_thread，更简洁
async def call_sync_simple():
    result = await asyncio.to_thread(blocking_function)
    return result

FastAPI中的异步使用

异步vs同步路由

from fastapi import FastAPI
import asyncio

app = FastAPI()

# ✅ 异步路由：处理I/O密集型任务
@app.get("/async-data")
async def get_async_data():
    await asyncio.sleep(1)
    return {"source": "async", "data": [1, 2, 3]}

# ✅ 同步路由：处理CPU密集型任务
@app.get("/sync-data")
def get_sync_data():
    result = sum(range(10**7))
    return {"source": "sync", "result": result}

异步HTTP请求（以httpx为例）

from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/fetch-multiple")
async def fetch_multiple():
    urls = [
        "https://api.github.com/users/octocat",
        "https://api.github.com/users/torvalds",
    ]
    async with httpx.AsyncClient(timeout=10) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
    return [r.json() for r in responses]

何时使用同步代码

用同步的场景

场景原因
CPU密集型任务（计算、加密、压缩）asyncio对CPU-bound无能为力，需丢线程池
无异步版本的库如Pillow、numpy计算部分
简单同步操作文件读写（os.path、open）、数学计算
启动/初始化代码app = FastAPI()这种只需跑一次

常见陷阱与避坑指南

陷阱1：忘记await

# ❌ 错误
async def bad():
    data = some_async_function()  # 返回协程对象，不会执行
    return data

# ✅ 正确
async def good():
    data = await some_async_function()
    return data

陷阱2：循环中串行await

# ❌ 错误：逐个等待，效率低
async def slow():
    results = []
    for url in urls:
        result = await fetch(url)
        results.append(result)

# ✅ 正确：并发等待
async def fast():
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)

实战：构建异步API服务

from fastapi import FastAPI, HTTPException
import httpx
import asyncio

app = FastAPI()

# 模拟异步数据库查询
async def query_users():
    await asyncio.sleep(0.1)
    return [
        {"id": 1, "name": "Alice"},
        {"id": 2, "name": "Bob"},
    ]

# 异步获取GitHub用户信息
async def fetch_github_user(username: str):
    async with httpx.AsyncClient() as client:
        resp = await client.get(f"https://api.github.com/users/{username}")
        if resp.status_code == 404:
            raise HTTPException(status_code=404, detail="User not found")
        resp.raise_for_status()
        return resp.json()

@app.get("/users")
async def list_users():
    """并发获取本地用户列表+GitHub公开信息"""
    users, octocat = await asyncio.gather(
        query_users(),
        fetch_github_user("octocat")
    )
    users.append({
        "id": octocat["id"],
        "name": octocat["name"],
        "is_github": True,
        "avatar": octocat["avatar_url"]
    })
    return users

性能优化建议

  1. 使用连接池：如httpx.AsyncClient、asyncpg.create_pool，避免频繁创建销毁连接
  2. 适当缓存：用lru_cache缓存同步计算，异步缓存可用asyncio_lru_cache
  3. 避免不必要的串行await：独立任务统一用asyncio.gather并发执行

总结

概念核心要点
async def定义协程函数，返回协程对象
await等待可等待对象，让出控制权
asyncio.sleep异步睡眠，不阻塞事件循环
asyncio.gather并发执行多个协程
asyncio.to_thread耗时同步操作丢线程池
FastAPI路由I/O密集用async，CPU密集用sync

💡 记住：异步的核心是处理大量并发I/O请求，但不要在异步函数中做CPU密集计算——那会让整个事件循环卡住。