Marico's space
用 FastAPI 写 API 几个月了，估计大家都被"async vs sync"这个话题折腾过。框架上手容易，两种写法都能跑，但真到线上选错了，那延迟、吞吐量、数据库驱动的坑一个都躲不掉。这篇把核心差异说透，附上压测数据，最后给个实操清单。
一句话总结——有可等待的 I/O（HTTP 调用、异步 DB 驱动、WebSocket）就用
async，CPU 密集型任务或者只能用老旧阻塞库的场景，老实用sync。
下面直接上干货。
FastAPI 底层跑的是 Starlette，再往下是 ASGI（异步服务器网关接口）。ASGI 服务器（比如 Uvicorn、Hypercorn）会起一个事件循环来调度协程。你把路由声明成 async def，FastAPI 就把这个协程注册到循环里；协程遇到 await 时，控制器交回给循环，其他请求继续跑。
from fastapi import FastAPI app = FastAPI() @app.get("/sync")
def sync_endpoint(): # 模拟阻塞 I/O
time.sleep(2) return {"msg": "sync done"} @app.get("/async")
async def async_endpoint(): # 模拟非阻塞 I/O
await asyncio.sleep(2) return {"msg": "async done"}
sync 版本里 time.sleep 把工作线程卡死，事件循环没法处理其他连接。async 版本用 asyncio.sleep 会主动让出控制权，单个 worker 就能扛住大量并发请求。
关键概念对照：
| 概念 | Async（async def） |
Sync（def） |
|---|---|---|
| 执行上下文 | 事件循环上的协程 | 线程池里的普通函数 |
await 使用 |
必须 await 可等待对象 | 不允许 |
| 并发模型 | 协作式多任务 | 抢占式多线程（用 ThreadPoolExecutor 时） |
| 典型场景 | I/O 密集型（DB、HTTP、WebSocket） | CPU 密集型或老旧阻塞库 |
如果你的端点对接的是异步兼容的服务——httpx.AsyncClient、databases 库、async SQLAlchemy 2.0、或者异步 Redis 客户端——果断用 async。事件循环能把大量 DB 查询交错执行，不用额外起线程。
from fastapi import FastAPI, Depends
import httpx app = FastAPI() async def fetch_user(user_id: int): async with httpx.AsyncClient() as client: r = await client.get(f"https://jsonplaceholder.typicode.com/users/{user_id}") r.raise_for_status() return r.json() @app.get("/users/{user_id}")
async def get_user(user_id: int): user = await fetch_user(user_id) return {"user": user}
Async 不会让 CPU 任务跑得更快。协程还是在同一个线程上跑，长时间 CPU 计算会把循环卡住，其他请求全得等着。这种情况要么保持 def 端点让 FastAPI 自动把它丢到线程池，要么显式用后台任务或 executor。
import hashlib def compute_hash(data: bytes) -> str: return hashlib.sha256(data).hexdigest() @app.post("/hash")
def hash_endpoint(payload: bytes): # FastAPI 会自动在线程池里跑这个
return {"hash": compute_hash(payload)}
有时候 async 和 sync 库混在一起用。路由保持 async，把阻塞调用包装进 run_in_threadpool：
from fastapi.concurrency import run_in_threadpool @app.get("/legacy")
async def legacy_endpoint(): result = await run_in_threadpool(some_blocking_library.do_work) return {"result": result}
如果你要部署到需要 WSGI 入口的 Serverless 平台（比如阿里云函数计算配 Mangum），只能走 sync 函数或者套一层兼容层。现在主流的容器（阿里云 ECS、腾讯云 TKE、Azure Container Instances）都原生支持 ASGI，async 完全没压力。
实测数据：Intel i7（2023 年款），Uvicorn 开 4 个 worker，端点分别 await asyncio.sleep(0.05)（async）和调用 time.sleep(0.05)（sync）。用 hey 压测，100 并发、总共 10000 请求，统计 RPS（每秒请求数）和平均延迟。
| 端点 | 模式 | RPS（≈） | 平均延迟（ms） |
|---|---|---|---|
/async-sleep |
async | 19,800 | 50 |
/sync-sleep |
sync | 4,200 | 240 |
/async-db（asyncpg） |
async | 12,500 | 80 |
/sync-db（psycopg2） |
sync | 6,800 | 150 |
| 因素 | Async | Sync |
|---|---|---|
| 复杂度 | 需要异步兼容库，调试更费劲 | 简单粗暴，什么库都能用 |
| 内存 | 线程少，内存占用低 | 线程多，内存压力大 |
| 第三方库支持 | 生态在成长但还有缺口（比如某些 ORM 扩展） | 成熟稳定，所有库都能跑 |
| 异常处理 | 异常沿事件循环冒泡传播，需要正确 await 才能正常抛出 | 直接 try/except 就完事 |
跑在低内存容器里选 async 有优势，绑死在老旧阻塞 ORM 上那就接受多几个线程的代价。
async def fetch(): return httpx.AsyncClient().get(url) # ❌ 返回协程对象，根本没执行
请求永远不会发出去，只会导致超时。切记——协程必须 await。
psycopg2（阻塞式）在 async 路由里会卡死整个循环。必须包装一下：
from fastapi.concurrency import run_in_threadpool @app.get("/users/{id}") async def get_user(id: int): row = await run_in_threadpool(lambda: sync_pg_query(id)) return row
关于这个问题的深入分析，可以看我们之前的文章《FastAPI 中修复 SQLAlchemy MissingGreenlet 错误的正确姿势》。
如果在 startup 事件里创建了一个全局的异步 DB 池，但忘了 await 创建过程，应用可能在池还没就绪时就开始接请求了，直接报 RuntimeError: Event loop is closed。《FastAPI Lifespan vs Startup 事件：Async 应用的正确打开方式》里有详细说明。
哪怕一个 for i in range(10**7): pass 都能把所有连接卡死。丢给线程池或者用 fastapi.BackgroundTasks 后台处理。
如果依赖 contextvars 存储请求级别的数据，注意 run_in_threadpool 会创建新线程，上下文不会自动带过去，需要手动传递。
import asyncpg
from fastapi import FastAPI, Depends app = FastAPI()
pool: asyncpg.Pool | None = None @app.on_event("startup")
async def create_pool(): global pool pool = await asyncpg.create_pool(dsn="postgresql://user:pwd@localhost/db") @app.on_event("shutdown")
async def close_pool(): await pool.close() async def get_conn() -> asyncpg.Connection: async with pool.acquire() as conn: yield conn @app.get("/items/{item_id}")
async def read_item(item_id: int, conn: asyncpg.Connection = Depends(get_conn)): row = await conn.fetchrow("SELECT * FROM items WHERE id = $1", item_id) return dict(row)
SQLAlchemy 2.0 原生支持异步 API，和 FastAPI 配合很顺滑。模式跟 asyncpg 类似，用 AsyncSession。
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi import Depends DATABASE_URL = "postgresql+asyncpg://user:pwd@localhost/db"
engine: AsyncEngine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async def get_async_session() -> AsyncSession: async with AsyncSessionLocal() as session: yield session @app.get("/users/{uid}")
async def get_user(uid: int, session: AsyncSession = Depends(get_async_session)): result = await session.execute(select(User).where(User.id == uid)) user = result.scalar_one_or_none() return user
实战经验：如果遇到 "MissingGreenlet" 报错，先检查是不是把同步
Session和异步代码混用了。《FastAPI 中修复 SQLAlchemy MissingGreenlet 错误的正确姿势》详细拆解了根因和修复方案。
长时间运行的 async 端点可能无意间占用 DB 连接不释放。用我们文章里介绍的监控工具记录连接检出时间，自动关闭空闲会话。《FastAPI SQLAlchemy Session 泄漏检测实战》有完整方案。
测试 async 路由用 pytest 配合 httpx.AsyncClient 就够了，关键是测试函数声明成 async，然后 await 客户端调用。
import pytest
from httpx import AsyncClient
from myapp.main import app # 假设你的 FastAPI 实例在这里
@pytest.mark.asyncio
async def test_async_endpoint(): async with AsyncClient(app=app, base_url="http://test") as ac: response = await ac.get("/async") assert response.status_code == 200 assert response.json() == {"msg": "async done"}
Sync 路由同样可以用这个异步客户端测——FastAPI 内部会把它丢到线程池执行，不需要单独写一套测试。
Mock 异步 DB 调用时，用 AsyncMock（Python 3.8+）来保持协程契约。
from unittest.mock