1. 项目背景
业务场景
"聚合报价服务"需要调用 3 个第三方 API（物流运费、支付手续费、汇率换算），然后计算出最终报价。小赵用最直观的方式实现：
@app.get("/quote")defget_quote(product_id:int):shipping=requests.get(f"https://api.shipping.com/calc?product={product_id}")# 800msfee=requests.get(f"https://api.payment.com/fee?product={product_id}")# 600msrate=requests.get(f"https://api.forex.com/rate?from=USD&to=CNY")# 400mstotal=shipping.json()["cost"]+fee.json()["fee"]+rate.json()["rate"]return{"total":total}接口响应时间：800 + 600 + 400 =1800ms。小赵想"FastAPI 不是号称高性能吗？怎么一个接口要 1.8 秒？"
他尝试把def改成async def：
@app.get("/quote")asyncdefget_quote(product_id:int):# 加了 asyncshipping=requests.get(...)# 还是同步 requests！...结果：还是 1.8 秒，而且并发 QPS 反而下降了。服务器 4 核 CPU，100 个并发请求，CPU 使用率只有 15%——因为所有协程都被阻塞在requests.get()上。
痛点
不掌握 Python 异步模型的核心原理，FastAPI 的高并发能力完全是无效的：
- 伪异步：
async def里面调同步requests.get()——协程阻塞，事件循环卡死，这是最典型的 FastAPI 性能陷阱。 - 串行等待：3 个 API 顺序调用，总耗时 = 最慢 API × 3。明明可以并发，却串行执行。
- 连接数爆炸：每次请求新建一个 HTTP 连接（三次握手 + TLS 握手），高并发下连接数超限。
- 超时失控：某个第三方 API 挂掉，接口 hang 住 30 秒才报错——线程池沾满，新请求排队等待。
FastAPI 是 ASGI 框架，它的高性能建立在async/await + 非阻塞 IO之上。不理解这个模型，就等于买了跑车但一直挂一档开。
2. 项目设计
场景：小赵在监控面板上看到报价接口 P99 延迟 3.2 秒。大师走过来，指着屏幕。
小胖：（震惊）“3.2 秒？！用户早关页面了。FastAPI 不是 Python 最快的框架吗？这跟 Flask 有区别吗？”
小白：“问题不在 FastAPI，在小赵的代码。你看第 1 章我们讲过——async def里的同步阻塞 IO（requests.get()）会卡住事件循环。但不止如此——他还串行调了 3 个 API。就像你去食堂打饭，先排队打饭、再排队打菜、再排队打汤——为什么不三个窗口一起排？”
大师："小白这个比喻好。今天我们把 Python 异步的三层概念讲透，大家以后写 FastAPI 就不会踩坑：
第一层——协程是什么：协程（coroutine）是一个可以在中途暂停和恢复的函数。Python 的async def定义协程，await是暂停点。暂停时，事件循环去执行其他协程。这就好比你在微波炉热饭的 3 分钟里，顺便去洗了个水果——而不是干等着微波炉叮。"
技术映射：Python 的
asyncio是基于事件循环的单线程并发模型。await点 = 协程交出控制权。当你在async def里调同步阻塞函数（如time.sleep(3)、requests.get()），控制权交不出去——事件循环被卡住，其他协程全部冻结。这叫做"协程的协作式调度"——你必须主动await。
小赵：“那我理解了——不能混用！async def里必须用异步库。但httpx.AsyncClient为什么就比requests.get()好在 async 环境里？”
小白：“requests.get()底层是同步 socket——socket.send()+socket.recv()，Python 线程在内核 I/O 上阻塞。而httpx.AsyncClient.get()是用asyncio的非阻塞 socket——当数据还没到达时，它立刻交还事件循环控制权，让其他协程继续执行。”
大师：“对。我再补一个容易忽略的细节——连接复用：”
# ❌ 串行 + 每次新建连接（慢）asyncdefbad():shipping=awaithttpx.AsyncClient().get(url1)# 新建连接（TCP+TLS）fee=awaithttpx.AsyncClient().get(url2)# 又新建连接rate=awaithttpx.AsyncClient().get(url3)# 又新建连接# ✓ 串行 + 连接复用（中）asyncdefbetter():asyncwithhttpx.AsyncClient()asclient:shipping=awaitclient.get(url1)fee=awaitclient.get(url2)rate=awaitclient.get(url3)# ✓✓ 并发 + 连接复用（快）：asyncio.gather 同时发起三个请求asyncdefbest():asyncwithhttpx.AsyncClient()asclient:shipping,fee,rate=awaitasyncio.gather(client.get(url1),client.get(url2),client.get(url3),)技术映射：
asyncio.gather()同时启动多个协程。总耗时 ≈ max(800ms, 600ms, 400ms) = 800ms——比串行的 1800ms 快了 2.25 倍。httpx.AsyncClient内部维护一个连接池，对同一 host 复用 TCP 连接，省去三次握手和 TLS 握手。
小胖：“那如果 3 个 API 有依赖怎么办——第二个 API 的请求参数依赖第一个 API 的返回值？”
大师：“那就是经典的’串行依赖’——没法并发。但可以优化：把独立的部分并发，依赖的部分串行。”
# 假设：报价需要运费+汇率，但汇率调用前需要先获取用户的国家代码asyncdefdependent():asyncwithhttpx.AsyncClient()asclient:# 并发：运费和用户信息可以同时查shipping,user_info=awaitasyncio.gather(client.get(shipping_url),client.get(user_url),)# 串行：汇率依赖用户的国家代码country=user_info.json()["country"]rate=awaitclient.get(f"https://api.forex.com/rate?country={country}")returnshipping.json()["cost"]+rate.json()["rate"]3. 项目实战——构建高性能报价服务
环境准备
pipinstallhttpx==0.27.0 pytest-asyncio==0.24.0分步实现
步骤一：搭建异步 HTTP 客户端（目标：连接复用 + 超时控制）
app/infrastructure/http_client.py：
importhttpxfromapp.core.configimportsettingsclassAsyncHTTPClient:"""异步 HTTP 客户端 —— 全局单例，连接池复用"""_instance:httpx.AsyncClient|None=None@classmethodasyncdefget_client(cls)->httpx.AsyncClient:ifcls._instanceisNone:cls._instance=httpx.AsyncClient(timeout=httpx.Timeout(connect=5.0,# TCP 连接超时read=10.0,# 读取响应超时write=5.0,# 发送请求超时pool=5.0,# 等待连接池可用连接超时),limits=httpx.Limits(max_keepalive_connections=20,# 最大保活连接数max_connections=50,# 总连接上限keepalive_expiry=30,# 保活时间（秒）),)returncls._instance@classmethodasyncdefclose(cls):ifcls._instance:awaitcls._instance.aclose()cls._instance=None步骤二：实现三种模式的报价服务（目标：直观对比性能差异）
app/domains/quote/service.py：
importtimeimportasyncioimporthttpxfromapp.infrastructure.http_clientimportAsyncHTTPClient# 模拟的第三方 API URL（实际环境需替换）SHIPPING_API="http://localhost:9001/shipping"PAYMENT_API="http://localhost:9002/payment-fee"FOREX_API="http://localhost:9003/forex-rate"classQuoteService:"""报价服务 —— 演示三种调用模式的性能差异"""# ═══════ 模式一：同步串行（最慢）═══defquote_sync_serial(self,product_id:int)->dict:"""同步串行：每个请求阻塞 0.5-1s"""start=time.perf_counter()resp1=httpx.get(f"{SHIPPING_API}?product={product_id}")# 阻塞resp2=httpx.get(f"{PAYMENT_API}?product={product_id}")# 阻塞resp3=httpx.get(FOREX_API)# 阻塞elapsed=time.perf_counter()-startreturn{"mode":"sync_serial","shipping":resp1.json().get("cost",0),"fee":resp2.json().get("fee",0),"rate":resp3.json().get("rate",0),"elapsed_ms":round(elapsed*1000,2),}# ═══════ 模式二：异步串行（快于同步，但未利用并发）═══asyncdefquote_async_serial(self,product_id:int)->dict:"""异步串行：非阻塞但顺序执行"""start=time.perf_counter()asyncwithhttpx.AsyncClient()asclient:resp1=awaitclient.get(f"{SHIPPING_API}?product={product_id}")resp2=awaitclient.get(f"{PAYMENT_API}?product={product_id}")resp3=awaitclient.get(FOREX_API)elapsed=time.perf_counter()-startreturn{"mode":"async_serial","shipping":resp1.json().get("cost",0),"fee":resp2.json().get("fee",0),"rate":resp3.json().get("rate",0),"elapsed_ms":round(elapsed*1000,2),}# ═══════ 模式三：异步并发（最快）═══asyncdefquote_async_concurrent(self,product_id:int)->dict:"""异步并发：三个请求同时发出，总耗时 = max(单个耗时)"""start=time.perf_counter()client=awaitAsyncHTTPClient.get_client()shipping_task=client.get(f"{SHIPPING_API}?product={product_id}")payment_task=client.get(f"{PAYMENT_API}?product={product_id}")forex_task=client.get(FOREX_API)# asyncio.gather 同时执行三个协程resp1,resp2,resp3=awaitasyncio.gather(shipping_task,payment_task,forex_task,# return_exceptions=True # 单个失败不影响其他)elapsed=time.perf_counter()-startreturn{"mode":"async_concurrent","shipping":resp1.json().get("cost",0),"fee":resp2.json().get("fee",0),"rate":resp3.json().get("rate",0),"elapsed_ms":round(elapsed*1000,2),}步骤三：增加并发控制（目标：使用 Semaphore 限制并发数）
classQuoteService:# ... 上面代码 ...# 信号量：限制同时调用第三方 API 的并发数_semaphore=asyncio.Semaphore(10)asyncdefquote_with_limit(self,product_id:int)->dict:"""带并发限制的报价——防止打爆第三方 API"""asyncwithself._semaphore:returnawaitself.quote_async_concurrent(product_id)步骤四：创建报价 API 路由（目标：在接口中对比三种模式）
app/domains/quote/api.py：
fromfastapiimportAPIRouter,Queryfromapp.domains.quote.serviceimportQuoteService router=APIRouter(prefix="/quote",tags=["报价服务"])quote_service=QuoteService()@router.get("/sync",summary="同步串行报价（慢）")defquote_sync(product_id:int=Query(...,gt=0)):"""def 端点 → 在线程池中执行，不阻塞事件循环"""return{"code":0,"data":quote_service.quote_sync_serial(product_id)}@router.get("/async-serial",summary="异步串行报价")asyncdefquote_async_serial(product_id:int=Query(...,gt=0)):return{"code":0,"data":awaitquote_service.quote_async_serial(product_id)}@router.get("/async-concurrent",summary="异步并发报价（推荐）")asyncdefquote_async_concurrent(product_id:int=Query(...,gt=0)):return{"code":0,"data":awaitquote_service.quote_async_concurrent(product_id)}步骤五：启动模拟服务并对比性能
# 启动三个模拟的第三方 APIpython scripts/mock_apis.py&# 起 3 个简单的 HTTP 服务（每个 500-1000ms 延迟）# 启动主服务uvicorn app.main:app--reload# ── 1. 同步串行 ──curl-shttp://localhost:8000/api/v1/quote/sync?product_id=1|python-mjson.tool# "elapsed_ms": 1850 ← 三个 API 延迟之和# ── 2. 异步串行 ──curl-shttp://localhost:8000/api/v1/quote/async-serial?product_id=1|python-mjson.tool# "elapsed_ms": 1800 ← 依然很慢（虽然非阻塞但顺序执行）# ── 3. 异步并发 ──curl-shttp://localhost:8000/api/v1/quote/async-concurrent?product_id=1|python-mjson.tool# "elapsed_ms": 620 ← 仅等于最慢的那个 API 延迟！# ── 4. 并发压测：比较 QPS ──# 同步模式 100 并发下 QPS ~50（线程池耗尽）# 异步并发模式 100 并发下 QPS ~800（事件循环充分利用）完整代码清单
本章完整代码见column/code/chapter17/，主要文件：
app/infrastructure/http_client.py：异步 HTTP 客户端app/domains/quote/service.py：三种模式的报价服务app/domains/quote/api.py：报价 API 路由
测试验证
importpytestimportasynciofromapp.domains.quote.serviceimportQuoteService@pytest.mark.asyncioasyncdeftest_async_concurrent_is_parallel():"""验证 asyncio.gather 真正实现了并发（总耗时 < 各任务之和）"""service=QuoteService()asyncdeffast_task():awaitasyncio.sleep(0.1)return"fast"asyncdefslow_task():awaitasyncio.sleep(0.3)return"slow"# 并发执行：总耗时应接近 max(0.1, 0.3) = 0.3sstart=asyncio.get_event_loop().time()results=awaitasyncio.gather(fast_task(),slow_task())elapsed=asyncio.get_event_loop().time()-startassertelapsed<0.35# 远小于 0.4（串行之和）assertresults==["fast","slow"]4. 项目总结
优点 & 缺点对比
| 模式 | async/await + asyncio.gather | 多线程 (ThreadPoolExecutor) | 多进程 | Node.js 事件循环 |
|---|---|---|---|---|
| IO 并发 | 优秀（协程切换零开销） | 中（线程切换有开销） | 低（进程切换开销大） | 优秀 |
| CPU 密集型 | 差（阻塞事件循环） | 中（受 GIL 限制） | 优秀 | 差 |
| 编程模型 | async/await（学习曲线中） | 同步代码 + 线程池 | 同步代码 | async/await |
| 内存占用 | 极低（一个协程 ~1KB） | 高（一个线程 ~8MB） | 极高 | 极低 |
适用场景
✓ 异步并发适用：
- 聚合多个下游 API 的 BFF（Backend for Frontend）接口
- 需要同时查询多个数据库/缓存的只读接口
- WebSocket 长连接管理
- 文件批量处理（并发读写多个文件）
- 微服务间批量调用
✗ 不适合异步：
- CPU 密集型计算（图片处理、加密解密）——用
def端点在独立线程池执行 - 只有单一数据源的简单 CRUD——async 带来的收益不明显
注意事项
- 不要混用同步库：
async def函数内不要调time.sleep()、requests.get()、同步数据库驱动。用asyncio.sleep()、httpx.AsyncClient、asyncpg。 asyncio.gather的 return_exceptions：默认False——任一协程异常，gather立即抛异常，其他协程被取消。设return_exceptions=True让单个失败不影响整体。- Semaphore 不是全局并发限制：
asyncio.Semaphore只限制当前事件循环内的并发。多 Worker 进程下需要 Redis 等外部计数器做全局限流。 - 连接池耗尽表现：大量
httpx.PoolTimeout异常。调大max_connections或增加keepalive_expiry加速连接回收。
常见踩坑经验
案例一：async def端点中的time.sleep()卡死事件循环
- 现象：100 并发请求，只有一个请求在执行，其余 99 个排队——QPS 只有 0.5。
- 根因：开发者在
async def函数中调了time.sleep(2)，事件循环被阻塞 2 秒。 - 解决：
await asyncio.sleep(2)或改用def端点让线程池处理。
案例二：asyncio.gather中一个任务挂起导致所有任务超时
- 现象：3 个 API 并发调用，其中一个超时 30s，其余两个 200ms 就返回了一直被拦住。
- 根因：
gather默认等待所有任务完成才返回。 - 解决：为每个任务单独设置 timeout —
asyncio.wait_for(task, timeout=5)；或使用asyncio.as_completed()先返回先处理。
案例三：httpx.AsyncClient提前关闭
- 现象：服务启动正常，运行几分钟后所有外部 API 调用报
RuntimeError: Event loop is closed。 - 根因：在 Lifespan 中创建了
AsyncClient，但在某次异常中没有正确关闭。下次请求时复用了一个半关闭的 client。 - 解决：在
app的 lifespan 事件中管理 client 的创建和关闭；或每次请求创建新的AsyncClient（性能略低但更安全）。
思考题
初级：修改报价服务，新增一个"超时兜底"模式——如果某个 API 在 1 秒内未响应，使用缓存中的上一次数据作为兜底（stale-while-revalidate 策略）。
进阶：如何使用
asyncio.TaskGroup（Python 3.11+）替代asyncio.gather？TaskGroup相比gather的优势是什么？提示：结构化并发。
答案提示：第 1 题使用
asyncio.wait_for(task, timeout=1)配合缓存。第 2 题：TaskGroup是 Python 的结构化并发原语——如果组内任一任务抛异常，所有子任务自动取消，不会出现"孤儿协程"。第 37 章深入事件循环诊断与性能极限。
延伸阅读与资源
NumPy 从入门到生产落地：全链路实战指南（科学计算/向量化）
Redis 8 实战精讲：从 CRUD 到源码，构建高可用缓存系统
Redis 实战修炼与原理进阶
Python 3实战精进：从脚本到高并发订单引擎
python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经
Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
MongoDB 实战进阶与内核修炼
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析
