1. 項目背景業務場景聚合報價服務需要調用 3 個第三方 API物流運費、支付手續費、匯率換算然后計算出最終報價。小趙用最直觀的方式實現app.get(/quote)defget_quote(product_id:int):shippingrequests.get(fhttps://api.shipping.com/calc?product{product_id})# 800msfeerequests.get(fhttps://api.payment.com/fee?product{product_id})# 600msraterequests.get(fhttps://api.forex.com/rate?fromUSDtoCNY)# 400mstotalshipping.json()[cost]fee.json()[fee]rate.json()[rate]return{total:total}接口響應時間800 600 400 1800ms。小趙想FastAPI 不是號稱高性能嗎怎么一個接口要 1.8 秒他嘗試把def改成async defapp.get(/quote)asyncdefget_quote(product_id:int):# 加了 asyncshippingrequests.get(...)# 還是同步 requests...結果還是 1.8 秒而且并發 QPS 反而下降了。服務器 4 核 CPU100 個并發請求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里的同步阻塞 IOrequests.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():shippingawaithttpx.AsyncClient().get(url1)# 新建連接TCPTLSfeeawaithttpx.AsyncClient().get(url2)# 又新建連接rateawaithttpx.AsyncClient().get(url3)# 又新建連接# ? 串行 連接復用中asyncdefbetter():asyncwithhttpx.AsyncClient()asclient:shippingawaitclient.get(url1)feeawaitclient.get(url2)rateawaitclient.get(url3)# ?? 并發 連接復用快asyncio.gather 同時發起三個請求asyncdefbest():asyncwithhttpx.AsyncClient()asclient:shipping,fee,rateawaitasyncio.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_infoawaitasyncio.gather(client.get(shipping_url),client.get(user_url),)# 串行匯率依賴用戶的國家代碼countryuser_info.json()[country]rateawaitclient.get(fhttps://api.forex.com/rate?country{country})returnshipping.json()[cost]rate.json()[rate]3. 項目實戰——構建高性能報價服務環境準備pipinstallhttpx0.27.0 pytest-asyncio0.24.0分步實現步驟一搭建異步 HTTP 客戶端目標連接復用 超時控制app/infrastructure/http_client.pyimporthttpxfromapp.core.configimportsettingsclassAsyncHTTPClient:異步 HTTP 客戶端 —— 全局單例連接池復用_instance:httpx.AsyncClient|NoneNoneclassmethodasyncdefget_client(cls)-httpx.AsyncClient:ifcls._instanceisNone:cls._instancehttpx.AsyncClient(timeouthttpx.Timeout(connect5.0,# TCP 連接超時read10.0,# 讀取響應超時write5.0,# 發送請求超時pool5.0,# 等待連接池可用連接超時),limitshttpx.Limits(max_keepalive_connections20,# 最大保活連接數max_connections50,# 總連接上限keepalive_expiry30,# 保活時間秒),)returncls._instanceclassmethodasyncdefclose(cls):ifcls._instance:awaitcls._instance.aclose()cls._instanceNone步驟二實現三種模式的報價服務目標直觀對比性能差異app/domains/quote/service.pyimporttimeimportasyncioimporthttpxfromapp.infrastructure.http_clientimportAsyncHTTPClient# 模擬的第三方 API URL實際環境需替換SHIPPING_APIhttp://localhost:9001/shippingPAYMENT_APIhttp://localhost:9002/payment-feeFOREX_APIhttp://localhost:9003/forex-rateclassQuoteService:報價服務 —— 演示三種調用模式的性能差異# ═══════ 模式一同步串行最慢═══defquote_sync_serial(self,product_id:int)-dict:同步串行每個請求阻塞 0.5-1sstarttime.perf_counter()resp1httpx.get(f{SHIPPING_API}?product{product_id})# 阻塞resp2httpx.get(f{PAYMENT_API}?product{product_id})# 阻塞resp3httpx.get(FOREX_API)# 阻塞elapsedtime.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:異步串行非阻塞但順序執行starttime.perf_counter()asyncwithhttpx.AsyncClient()asclient:resp1awaitclient.get(f{SHIPPING_API}?product{product_id})resp2awaitclient.get(f{PAYMENT_API}?product{product_id})resp3awaitclient.get(FOREX_API)elapsedtime.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(單個耗時)starttime.perf_counter()clientawaitAsyncHTTPClient.get_client()shipping_taskclient.get(f{SHIPPING_API}?product{product_id})payment_taskclient.get(f{PAYMENT_API}?product{product_id})forex_taskclient.get(FOREX_API)# asyncio.gather 同時執行三個協程resp1,resp2,resp3awaitasyncio.gather(shipping_task,payment_task,forex_task,# return_exceptionsTrue # 單個失敗不影響其他)elapsedtime.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 的并發數_semaphoreasyncio.Semaphore(10)asyncdefquote_with_limit(self,product_id:int)-dict:帶并發限制的報價——防止打爆第三方 APIasyncwithself._semaphore:returnawaitself.quote_async_concurrent(product_id)步驟四創建報價 API 路由目標在接口中對比三種模式app/domains/quote/api.pyfromfastapiimportAPIRouter,Queryfromapp.domains.quote.serviceimportQuoteService routerAPIRouter(prefix/quote,tags[報價服務])quote_serviceQuoteService()router.get(/sync,summary同步串行報價慢)defquote_sync(product_id:intQuery(...,gt0)):def 端點 → 在線程池中執行不阻塞事件循環return{code:0,data:quote_service.quote_sync_serial(product_id)}router.get(/async-serial,summary異步串行報價)asyncdefquote_async_serial(product_id:intQuery(...,gt0)):return{code:0,data:awaitquote_service.quote_async_serial(product_id)}router.get(/async-concurrent,summary異步并發報價推薦)asyncdefquote_async_concurrent(product_id:intQuery(...,gt0)):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_id1|python-mjson.tool# elapsed_ms: 1850 ← 三個 API 延遲之和# ── 2. 異步串行 ──curl-shttp://localhost:8000/api/v1/quote/async-serial?product_id1|python-mjson.tool# elapsed_ms: 1800 ← 依然很慢雖然非阻塞但順序執行# ── 3. 異步并發 ──curl-shttp://localhost:8000/api/v1/quote/async-concurrent?product_id1|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.serviceimportQuoteServicepytest.mark.asyncioasyncdeftest_async_concurrent_is_parallel():驗證 asyncio.gather 真正實現了并發總耗時 各任務之和serviceQuoteService()asyncdeffast_task():awaitasyncio.sleep(0.1)returnfastasyncdefslow_task():awaitasyncio.sleep(0.3)returnslow# 并發執行總耗時應接近 max(0.1, 0.3) 0.3sstartasyncio.get_event_loop().time()resultsawaitasyncio.gather(fast_task(),slow_task())elapsedasyncio.get_event_loop().time()-startassertelapsed0.35# 遠小于 0.4串行之和assertresults[fast,slow]4. 項目總結優點 缺點對比模式async/await asyncio.gather多線程 (ThreadPoolExecutor)多進程Node.js 事件循環IO 并發優秀協程切換零開銷中線程切換有開銷低進程切換開銷大優秀CPU 密集型差阻塞事件循環中受 GIL 限制優秀差編程模型async/await學習曲線中同步代碼 線程池同步代碼async/await內存占用極低一個協程 ~1KB高一個線程 ~8MB極高極低適用場景? 異步并發適用聚合多個下游 API 的 BFFBackend 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_exceptionsTrue讓單個失敗不影響整體。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, timeout5)或使用asyncio.as_completed()先返回先處理。案例三httpx.AsyncClient提前關閉現象服務啟動正常運行幾分鐘后所有外部 API 調用報RuntimeError: Event loop is closed。根因在 Lifespan 中創建了AsyncClient但在某次異常中沒有正確關閉。下次請求時復用了一個半關閉的 client。解決在app的 lifespan 事件中管理 client 的創建和關閉或每次請求創建新的AsyncClient性能略低但更安全。思考題初級修改報價服務新增一個超時兜底模式——如果某個 API 在 1 秒內未響應使用緩存中的上一次數據作為兜底stale-while-revalidate 策略。進階如何使用asyncio.TaskGroupPython 3.11替代asyncio.gatherTaskGroup相比gather的優勢是什么提示結構化并發。答案提示第 1 題使用asyncio.wait_for(task, timeout1)配合緩存。第 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 實戰修煉與源碼剖析