)
1. 項目背景業(yè)務(wù)場景聚合報價服務(wù)需要調(diào)用 3 個第三方 API物流運費、支付手續(xù)費、匯率換算然后計算出最終報價。小趙用最直觀的方式實現(xiàn)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}接口響應(yīng)時間800 600 400 1800ms。小趙想FastAPI 不是號稱高性能嗎怎么一個接口要 1.8 秒他嘗試把def改成async defapp.get(/quote)asyncdefget_quote(product_id:int):# 加了 asyncshippingrequests.get(...)# 還是同步 requests...結(jié)果還是 1.8 秒而且并發(fā) QPS 反而下降了。服務(wù)器 4 核 CPU100 個并發(fā)請求CPU 使用率只有 15%——因為所有協(xié)程都被阻塞在requests.get()上。痛點不掌握 Python 異步模型的核心原理FastAPI 的高并發(fā)能力完全是無效的偽異步async def里面調(diào)同步requests.get()——協(xié)程阻塞事件循環(huán)卡死這是最典型的 FastAPI 性能陷阱。串行等待3 個 API 順序調(diào)用總耗時 最慢 API × 3。明明可以并發(fā)卻串行執(zhí)行。連接數(shù)爆炸每次請求新建一個 HTTP 連接三次握手 TLS 握手高并發(fā)下連接數(shù)超限。超時失控某個第三方 API 掛掉接口 hang 住 30 秒才報錯——線程池沾滿新請求排隊等待。FastAPI 是 ASGI 框架它的高性能建立在async/await 非阻塞 IO之上。不理解這個模型就等于買了跑車但一直掛一檔開。2. 項目設(shè)計場景小趙在監(jiān)控面板上看到報價接口 P99 延遲 3.2 秒。大師走過來指著屏幕。小胖震驚“3.2 秒用戶早關(guān)頁面了。FastAPI 不是 Python 最快的框架嗎這跟 Flask 有區(qū)別嗎”小白“問題不在 FastAPI在小趙的代碼。你看第 1 章我們講過——async def里的同步阻塞 IOrequests.get()會卡住事件循環(huán)。但不止如此——他還串行調(diào)了 3 個 API。就像你去食堂打飯先排隊打飯、再排隊打菜、再排隊打湯——為什么不三個窗口一起排”大師小白這個比喻好。今天我們把 Python 異步的三層概念講透大家以后寫 FastAPI 就不會踩坑第一層——協(xié)程是什么協(xié)程coroutine是一個可以在中途暫停和恢復(fù)的函數(shù)。Python 的async def定義協(xié)程await是暫停點。暫停時事件循環(huán)去執(zhí)行其他協(xié)程。這就好比你在微波爐熱飯的 3 分鐘里順便去洗了個水果——而不是干等著微波爐叮。技術(shù)映射Python 的asyncio是基于事件循環(huán)的單線程并發(fā)模型。await點 協(xié)程交出控制權(quán)。當你在async def里調(diào)同步阻塞函數(shù)如time.sleep(3)、requests.get()控制權(quán)交不出去——事件循環(huán)被卡住其他協(xié)程全部凍結(jié)。這叫做協(xié)程的協(xié)作式調(diào)度——你必須主動await。小趙“那我理解了——不能混用async def里必須用異步庫。但httpx.AsyncClient為什么就比requests.get()好在 async 環(huán)境里”小白“requests.get()底層是同步 socket——socket.send()socket.recv()Python 線程在內(nèi)核 I/O 上阻塞。而httpx.AsyncClient.get()是用asyncio的非阻塞 socket——當數(shù)據(jù)還沒到達時它立刻交還事件循環(huán)控制權(quán)讓其他協(xié)程繼續(xù)執(zhí)行?！贝髱煛皩?。我再補一個容易忽略的細節(jié)——連接復(fù)用”# ? 串行 每次新建連接慢asyncdefbad():shippingawaithttpx.AsyncClient().get(url1)# 新建連接TCPTLSfeeawaithttpx.AsyncClient().get(url2)# 又新建連接rateawaithttpx.AsyncClient().get(url3)# 又新建連接# ? 串行 連接復(fù)用中asyncdefbetter():asyncwithhttpx.AsyncClient()asclient:shippingawaitclient.get(url1)feeawaitclient.get(url2)rateawaitclient.get(url3)# ?? 并發(fā) 連接復(fù)用快asyncio.gather 同時發(fā)起三個請求asyncdefbest():asyncwithhttpx.AsyncClient()asclient:shipping,fee,rateawaitasyncio.gather(client.get(url1),client.get(url2),client.get(url3),)技術(shù)映射asyncio.gather()同時啟動多個協(xié)程?？偤臅r ≈ max(800ms, 600ms, 400ms) 800ms——比串行的 1800ms 快了 2.25 倍。httpx.AsyncClient內(nèi)部維護一個連接池對同一 host 復(fù)用 TCP 連接省去三次握手和 TLS 握手。小胖“那如果 3 個 API 有依賴怎么辦——第二個 API 的請求參數(shù)依賴第一個 API 的返回值”大師“那就是經(jīng)典的’串行依賴’——沒法并發(fā)。但可以優(yōu)化把獨立的部分并發(fā)依賴的部分串行。”# 假設(shè)報價需要運費匯率但匯率調(diào)用前需要先獲取用戶的國家代碼asyncdefdependent():asyncwithhttpx.AsyncClient()asclient:# 并發(fā)運費和用戶信息可以同時查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. 項目實戰(zhàn)——構(gòu)建高性能報價服務(wù)環(huán)境準備pipinstallhttpx0.27.0 pytest-asyncio0.24.0分步實現(xiàn)步驟一搭建異步 HTTP 客戶端目標連接復(fù)用 超時控制app/infrastructure/http_client.pyimporthttpxfromapp.core.configimportsettingsclassAsyncHTTPClient:異步 HTTP 客戶端 —— 全局單例連接池復(fù)用_instance:httpx.AsyncClient|NoneNoneclassmethodasyncdefget_client(cls)-httpx.AsyncClient:ifcls._instanceisNone:cls._instancehttpx.AsyncClient(timeouthttpx.Timeout(connect5.0,# TCP 連接超時read10.0,# 讀取響應(yīng)超時write5.0,# 發(fā)送請求超時pool5.0,# 等待連接池可用連接超時),limitshttpx.Limits(max_keepalive_connections20,# 最大?；钸B接數(shù)max_connections50,# 總連接上限keepalive_expiry30,# ?；顣r間秒),)returncls._instanceclassmethodasyncdefclose(cls):ifcls._instance:awaitcls._instance.aclose()cls._instanceNone步驟二實現(xiàn)三種模式的報價服務(wù)目標直觀對比性能差異app/domains/quote/service.pyimporttimeimportasyncioimporthttpxfromapp.infrastructure.http_clientimportAsyncHTTPClient# 模擬的第三方 API URL實際環(huán)境需替換SHIPPING_APIhttp://localhost:9001/shippingPAYMENT_APIhttp://localhost:9002/payment-feeFOREX_APIhttp://localhost:9003/forex-rateclassQuoteService:報價服務(wù) —— 演示三種調(diào)用模式的性能差異# ═══════ 模式一同步串行最慢═══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),}# ═══════ 模式二異步串行快于同步但未利用并發(fā)═══asyncdefquote_async_serial(self,product_id:int)-dict:異步串行非阻塞但順序執(zhí)行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),}# ═══════ 模式三異步并發(fā)最快═══asyncdefquote_async_concurrent(self,product_id:int)-dict:異步并發(fā)三個請求同時發(fā)出總耗時 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 同時執(zhí)行三個協(xié)程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),}步驟三增加并發(fā)控制目標使用 Semaphore 限制并發(fā)數(shù)classQuoteService:# ... 上面代碼 ...# 信號量限制同時調(diào)用第三方 API 的并發(fā)數(shù)_semaphoreasyncio.Semaphore(10)asyncdefquote_with_limit(self,product_id:int)-dict:帶并發(fā)限制的報價——防止打爆第三方 APIasyncwithself._semaphore:returnawaitself.quote_async_concurrent(product_id)步驟四創(chuàng)建報價 API 路由目標在接口中對比三種模式app/domains/quote/api.pyfromfastapiimportAPIRouter,Queryfromapp.domains.quote.serviceimportQuoteService routerAPIRouter(prefix/quote,tags[報價服務(wù)])quote_serviceQuoteService()router.get(/sync,summary同步串行報價慢)defquote_sync(product_id:intQuery(...,gt0)):def 端點 → 在線程池中執(zhí)行不阻塞事件循環(huán)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異步并發(fā)報價推薦)asyncdefquote_async_concurrent(product_id:intQuery(...,gt0)):return{code:0,data:awaitquote_service.quote_async_concurrent(product_id)}步驟五啟動模擬服務(wù)并對比性能# 啟動三個模擬的第三方 APIpython scripts/mock_apis.py# 起 3 個簡單的 HTTP 服務(wù)每個 500-1000ms 延遲# 啟動主服務(wù)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 ← 依然很慢雖然非阻塞但順序執(zhí)行# ── 3. 異步并發(fā) ──curl-shttp://localhost:8000/api/v1/quote/async-concurrent?product_id1|python-mjson.tool# elapsed_ms: 620 ← 僅等于最慢的那個 API 延遲# ── 4. 并發(fā)壓測比較 QPS ──# 同步模式 100 并發(fā)下 QPS ~50線程池耗盡# 異步并發(fā)模式 100 并發(fā)下 QPS ~800事件循環(huán)充分利用完整代碼清單本章完整代碼見column/code/chapter17/主要文件app/infrastructure/http_client.py異步 HTTP 客戶端app/domains/quote/service.py三種模式的報價服務(wù)app/domains/quote/api.py報價 API 路由測試驗證importpytestimportasynciofromapp.domains.quote.serviceimportQuoteServicepytest.mark.asyncioasyncdeftest_async_concurrent_is_parallel():驗證 asyncio.gather 真正實現(xiàn)了并發(fā)總耗時 各任務(wù)之和serviceQuoteService()asyncdeffast_task():awaitasyncio.sleep(0.1)returnfastasyncdefslow_task():awaitasyncio.sleep(0.3)returnslow# 并發(fā)執(zhí)行總耗時應(yīng)接近 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. 項目總結(jié)優(yōu)點 缺點對比模式async/await asyncio.gather多線程 (ThreadPoolExecutor)多進程Node.js 事件循環(huán)IO 并發(fā)優(yōu)秀協(xié)程切換零開銷中線程切換有開銷低進程切換開銷大優(yōu)秀CPU 密集型差阻塞事件循環(huán)中受 GIL 限制優(yōu)秀差編程模型async/await學(xué)習(xí)曲線中同步代碼 線程池同步代碼async/await內(nèi)存占用極低一個協(xié)程 ~1KB高一個線程 ~8MB極高極低適用場景? 異步并發(fā)適用聚合多個下游 API 的 BFFBackend for Frontend接口需要同時查詢多個數(shù)據(jù)庫/緩存的只讀接口WebSocket 長連接管理文件批量處理并發(fā)讀寫多個文件微服務(wù)間批量調(diào)用? 不適合異步CPU 密集型計算圖片處理、加密解密——用def端點在獨立線程池執(zhí)行只有單一數(shù)據(jù)源的簡單 CRUD——async 帶來的收益不明顯注意事項不要混用同步庫async def函數(shù)內(nèi)不要調(diào)time.sleep()、requests.get()、同步數(shù)據(jù)庫驅(qū)動。用asyncio.sleep()、httpx.AsyncClient、asyncpg。asyncio.gather的 return_exceptions默認False——任一協(xié)程異常gather立即拋異常其他協(xié)程被取消。設(shè)return_exceptionsTrue讓單個失敗不影響整體。Semaphore 不是全局并發(fā)限制asyncio.Semaphore只限制當前事件循環(huán)內(nèi)的并發(fā)。多 Worker 進程下需要 Redis 等外部計數(shù)器做全局限流。連接池耗盡表現(xiàn)大量httpx.PoolTimeout異常。調(diào)大max_connections或增加keepalive_expiry加速連接回收。常見踩坑經(jīng)驗案例一async def端點中的time.sleep()卡死事件循環(huán)現(xiàn)象100 并發(fā)請求只有一個請求在執(zhí)行其余 99 個排隊——QPS 只有 0.5。根因開發(fā)者在async def函數(shù)中調(diào)了time.sleep(2)事件循環(huán)被阻塞 2 秒。解決await asyncio.sleep(2)或改用def端點讓線程池處理。案例二asyncio.gather中一個任務(wù)掛起導(dǎo)致所有任務(wù)超時現(xiàn)象3 個 API 并發(fā)調(diào)用其中一個超時 30s其余兩個 200ms 就返回了一直被攔住。根因gather默認等待所有任務(wù)完成才返回。解決為每個任務(wù)單獨設(shè)置 timeout —asyncio.wait_for(task, timeout5)或使用asyncio.as_completed()先返回先處理。案例三httpx.AsyncClient提前關(guān)閉現(xiàn)象服務(wù)啟動正常運行幾分鐘后所有外部 API 調(diào)用報RuntimeError: Event loop is closed。根因在 Lifespan 中創(chuàng)建了AsyncClient但在某次異常中沒有正確關(guān)閉。下次請求時復(fù)用了一個半關(guān)閉的 client。解決在app的 lifespan 事件中管理 client 的創(chuàng)建和關(guān)閉或每次請求創(chuàng)建新的AsyncClient性能略低但更安全。思考題初級修改報價服務(wù)新增一個超時兜底模式——如果某個 API 在 1 秒內(nèi)未響應(yīng)使用緩存中的上一次數(shù)據(jù)作為兜底stale-while-revalidate 策略。進階如何使用asyncio.TaskGroupPython 3.11替代asyncio.gatherTaskGroup相比gather的優(yōu)勢是什么提示結(jié)構(gòu)化并發(fā)。答案提示第 1 題使用asyncio.wait_for(task, timeout1)配合緩存。第 2 題TaskGroup是 Python 的結(jié)構(gòu)化并發(fā)原語——如果組內(nèi)任一任務(wù)拋異常所有子任務(wù)自動取消不會出現(xiàn)孤兒協(xié)程。第 37 章深入事件循環(huán)診斷與性能極限。延伸閱讀與資源NumPy 從入門到生產(chǎn)落地全鏈路實戰(zhàn)指南科學(xué)計算/向量化Redis 8 實戰(zhàn)精講從 CRUD 到源碼構(gòu)建高可用緩存系統(tǒng)Redis 實戰(zhàn)修煉與原理進階Python 3實戰(zhàn)精進從腳本到高并發(fā)訂單引擎python入門Rquests從菜鳥腳本到企業(yè)級SDK的網(wǎng)絡(luò)實戰(zhàn)圣經(jīng)Milvus向量數(shù)據(jù)庫實戰(zhàn)修煉從 0 到 1精通向量檢索與生產(chǎn)落地MongoDB 實戰(zhàn)進階與內(nèi)核修煉后端工程師的 AI 轉(zhuǎn)型第一課Ollama 與私有化大模型實戰(zhàn)10倍開發(fā)者的 Dify 魔法書從零構(gòu)建全棧 AI 應(yīng)用后端工程師轉(zhuǎn)型AI第一課-Ollama 與私有化大模型實戰(zhàn)大型語言模型(LLM) vLLM 高性能推理落地實戰(zhàn)Agent開發(fā)之LlamaIndex 實戰(zhàn)修煉與源碼進階大語言模型Transformers 實戰(zhàn)修煉與源碼剖析 第17章：FastAPI異步編程與非阻塞 IO 實戰(zhàn)
1. 項目背景業(yè)務(wù)場景聚合報價服務(wù)需要調(diào)用 3 個第三方 API物流運費、支付手續(xù)費、匯率換算然后計算出最終報價。小趙用最直觀的方式實現(xiàn)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}接口響應(yīng)時間800 600 400 1800ms。小趙想FastAPI 不是號稱高性能嗎怎么一個接口要 1.8 秒他嘗試把def改成async defapp.get(/quote)asyncdefget_quote(product_id:int):# 加了 asyncshippingrequests.get(...)# 還是同步 requests...結(jié)果還是 1.8 秒而且并發(fā) QPS 反而下降了。服務(wù)器 4 核 CPU100 個并發(fā)請求CPU 使用率只有 15%——因為所有協(xié)程都被阻塞在requests.get()上。痛點不掌握 Python 異步模型的核心原理FastAPI 的高并發(fā)能力完全是無效的偽異步async def里面調(diào)同步requests.get()——協(xié)程阻塞事件循環(huán)卡死這是最典型的 FastAPI 性能陷阱。串行等待3 個 API 順序調(diào)用總耗時 最慢 API × 3。明明可以并發(fā)卻串行執(zhí)行。連接數(shù)爆炸每次請求新建一個 HTTP 連接三次握手 TLS 握手高并發(fā)下連接數(shù)超限。超時失控某個第三方 API 掛掉接口 hang 住 30 秒才報錯——線程池沾滿新請求排隊等待。FastAPI 是 ASGI 框架它的高性能建立在async/await 非阻塞 IO之上。不理解這個模型就等于買了跑車但一直掛一檔開。2. 項目設(shè)計場景小趙在監(jiān)控面板上看到報價接口 P99 延遲 3.2 秒。大師走過來指著屏幕。小胖震驚“3.2 秒用戶早關(guān)頁面了。FastAPI 不是 Python 最快的框架嗎這跟 Flask 有區(qū)別嗎”小白“問題不在 FastAPI在小趙的代碼。你看第 1 章我們講過——async def里的同步阻塞 IOrequests.get()會卡住事件循環(huán)。但不止如此——他還串行調(diào)了 3 個 API。就像你去食堂打飯先排隊打飯、再排隊打菜、再排隊打湯——為什么不三個窗口一起排”大師小白這個比喻好。今天我們把 Python 異步的三層概念講透大家以后寫 FastAPI 就不會踩坑第一層——協(xié)程是什么協(xié)程coroutine是一個可以在中途暫停和恢復(fù)的函數(shù)。Python 的async def定義協(xié)程await是暫停點。暫停時事件循環(huán)去執(zhí)行其他協(xié)程。這就好比你在微波爐熱飯的 3 分鐘里順便去洗了個水果——而不是干等著微波爐叮。技術(shù)映射Python 的asyncio是基于事件循環(huán)的單線程并發(fā)模型。await點 協(xié)程交出控制權(quán)。當你在async def里調(diào)同步阻塞函數(shù)如time.sleep(3)、requests.get()控制權(quán)交不出去——事件循環(huán)被卡住其他協(xié)程全部凍結(jié)。這叫做協(xié)程的協(xié)作式調(diào)度——你必須主動await。小趙“那我理解了——不能混用async def里必須用異步庫。但httpx.AsyncClient為什么就比requests.get()好在 async 環(huán)境里”小白“requests.get()底層是同步 socket——socket.send()socket.recv()Python 線程在內(nèi)核 I/O 上阻塞。而httpx.AsyncClient.get()是用asyncio的非阻塞 socket——當數(shù)據(jù)還沒到達時它立刻交還事件循環(huán)控制權(quán)讓其他協(xié)程繼續(xù)執(zhí)行?！贝髱煛皩?。我再補一個容易忽略的細節(jié)——連接復(fù)用”# ? 串行 每次新建連接慢asyncdefbad():shippingawaithttpx.AsyncClient().get(url1)# 新建連接TCPTLSfeeawaithttpx.AsyncClient().get(url2)# 又新建連接rateawaithttpx.AsyncClient().get(url3)# 又新建連接# ? 串行 連接復(fù)用中asyncdefbetter():asyncwithhttpx.AsyncClient()asclient:shippingawaitclient.get(url1)feeawaitclient.get(url2)rateawaitclient.get(url3)# ?? 并發(fā) 連接復(fù)用快asyncio.gather 同時發(fā)起三個請求asyncdefbest():asyncwithhttpx.AsyncClient()asclient:shipping,fee,rateawaitasyncio.gather(client.get(url1),client.get(url2),client.get(url3),)技術(shù)映射asyncio.gather()同時啟動多個協(xié)程?？偤臅r ≈ max(800ms, 600ms, 400ms) 800ms——比串行的 1800ms 快了 2.25 倍。httpx.AsyncClient內(nèi)部維護一個連接池對同一 host 復(fù)用 TCP 連接省去三次握手和 TLS 握手。小胖“那如果 3 個 API 有依賴怎么辦——第二個 API 的請求參數(shù)依賴第一個 API 的返回值”大師“那就是經(jīng)典的’串行依賴’——沒法并發(fā)。但可以優(yōu)化把獨立的部分并發(fā)依賴的部分串行。”# 假設(shè)報價需要運費匯率但匯率調(diào)用前需要先獲取用戶的國家代碼asyncdefdependent():asyncwithhttpx.AsyncClient()asclient:# 并發(fā)運費和用戶信息可以同時查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. 項目實戰(zhàn)——構(gòu)建高性能報價服務(wù)環(huán)境準備pipinstallhttpx0.27.0 pytest-asyncio0.24.0分步實現(xiàn)步驟一搭建異步 HTTP 客戶端目標連接復(fù)用 超時控制app/infrastructure/http_client.pyimporthttpxfromapp.core.configimportsettingsclassAsyncHTTPClient:異步 HTTP 客戶端 —— 全局單例連接池復(fù)用_instance:httpx.AsyncClient|NoneNoneclassmethodasyncdefget_client(cls)-httpx.AsyncClient:ifcls._instanceisNone:cls._instancehttpx.AsyncClient(timeouthttpx.Timeout(connect5.0,# TCP 連接超時read10.0,# 讀取響應(yīng)超時write5.0,# 發(fā)送請求超時pool5.0,# 等待連接池可用連接超時),limitshttpx.Limits(max_keepalive_connections20,# 最大?；钸B接數(shù)max_connections50,# 總連接上限keepalive_expiry30,# ?；顣r間秒),)returncls._instanceclassmethodasyncdefclose(cls):ifcls._instance:awaitcls._instance.aclose()cls._instanceNone步驟二實現(xiàn)三種模式的報價服務(wù)目標直觀對比性能差異app/domains/quote/service.pyimporttimeimportasyncioimporthttpxfromapp.infrastructure.http_clientimportAsyncHTTPClient# 模擬的第三方 API URL實際環(huán)境需替換SHIPPING_APIhttp://localhost:9001/shippingPAYMENT_APIhttp://localhost:9002/payment-feeFOREX_APIhttp://localhost:9003/forex-rateclassQuoteService:報價服務(wù) —— 演示三種調(diào)用模式的性能差異# ═══════ 模式一同步串行最慢═══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),}# ═══════ 模式二異步串行快于同步但未利用并發(fā)═══asyncdefquote_async_serial(self,product_id:int)-dict:異步串行非阻塞但順序執(zhí)行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),}# ═══════ 模式三異步并發(fā)最快═══asyncdefquote_async_concurrent(self,product_id:int)-dict:異步并發(fā)三個請求同時發(fā)出總耗時 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 同時執(zhí)行三個協(xié)程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),}步驟三增加并發(fā)控制目標使用 Semaphore 限制并發(fā)數(shù)classQuoteService:# ... 上面代碼 ...# 信號量限制同時調(diào)用第三方 API 的并發(fā)數(shù)_semaphoreasyncio.Semaphore(10)asyncdefquote_with_limit(self,product_id:int)-dict:帶并發(fā)限制的報價——防止打爆第三方 APIasyncwithself._semaphore:returnawaitself.quote_async_concurrent(product_id)步驟四創(chuàng)建報價 API 路由目標在接口中對比三種模式app/domains/quote/api.pyfromfastapiimportAPIRouter,Queryfromapp.domains.quote.serviceimportQuoteService routerAPIRouter(prefix/quote,tags[報價服務(wù)])quote_serviceQuoteService()router.get(/sync,summary同步串行報價慢)defquote_sync(product_id:intQuery(...,gt0)):def 端點 → 在線程池中執(zhí)行不阻塞事件循環(huán)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異步并發(fā)報價推薦)asyncdefquote_async_concurrent(product_id:intQuery(...,gt0)):return{code:0,data:awaitquote_service.quote_async_concurrent(product_id)}步驟五啟動模擬服務(wù)并對比性能# 啟動三個模擬的第三方 APIpython scripts/mock_apis.py# 起 3 個簡單的 HTTP 服務(wù)每個 500-1000ms 延遲# 啟動主服務(wù)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 ← 依然很慢雖然非阻塞但順序執(zhí)行# ── 3. 異步并發(fā) ──curl-shttp://localhost:8000/api/v1/quote/async-concurrent?product_id1|python-mjson.tool# elapsed_ms: 620 ← 僅等于最慢的那個 API 延遲# ── 4. 并發(fā)壓測比較 QPS ──# 同步模式 100 并發(fā)下 QPS ~50線程池耗盡# 異步并發(fā)模式 100 并發(fā)下 QPS ~800事件循環(huán)充分利用完整代碼清單本章完整代碼見column/code/chapter17/主要文件app/infrastructure/http_client.py異步 HTTP 客戶端app/domains/quote/service.py三種模式的報價服務(wù)app/domains/quote/api.py報價 API 路由測試驗證importpytestimportasynciofromapp.domains.quote.serviceimportQuoteServicepytest.mark.asyncioasyncdeftest_async_concurrent_is_parallel():驗證 asyncio.gather 真正實現(xiàn)了并發(fā)總耗時 各任務(wù)之和serviceQuoteService()asyncdeffast_task():awaitasyncio.sleep(0.1)returnfastasyncdefslow_task():awaitasyncio.sleep(0.3)returnslow# 并發(fā)執(zhí)行總耗時應(yīng)接近 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. 項目總結(jié)優(yōu)點 缺點對比模式async/await asyncio.gather多線程 (ThreadPoolExecutor)多進程Node.js 事件循環(huán)IO 并發(fā)優(yōu)秀協(xié)程切換零開銷中線程切換有開銷低進程切換開銷大優(yōu)秀CPU 密集型差阻塞事件循環(huán)中受 GIL 限制優(yōu)秀差編程模型async/await學(xué)習(xí)曲線中同步代碼 線程池同步代碼async/await內(nèi)存占用極低一個協(xié)程 ~1KB高一個線程 ~8MB極高極低適用場景? 異步并發(fā)適用聚合多個下游 API 的 BFFBackend for Frontend接口需要同時查詢多個數(shù)據(jù)庫/緩存的只讀接口WebSocket 長連接管理文件批量處理并發(fā)讀寫多個文件微服務(wù)間批量調(diào)用? 不適合異步CPU 密集型計算圖片處理、加密解密——用def端點在獨立線程池執(zhí)行只有單一數(shù)據(jù)源的簡單 CRUD——async 帶來的收益不明顯注意事項不要混用同步庫async def函數(shù)內(nèi)不要調(diào)time.sleep()、requests.get()、同步數(shù)據(jù)庫驅(qū)動。用asyncio.sleep()、httpx.AsyncClient、asyncpg。asyncio.gather的 return_exceptions默認False——任一協(xié)程異常gather立即拋異常其他協(xié)程被取消。設(shè)return_exceptionsTrue讓單個失敗不影響整體。Semaphore 不是全局并發(fā)限制asyncio.Semaphore只限制當前事件循環(huán)內(nèi)的并發(fā)。多 Worker 進程下需要 Redis 等外部計數(shù)器做全局限流。連接池耗盡表現(xiàn)大量httpx.PoolTimeout異常。調(diào)大max_connections或增加keepalive_expiry加速連接回收。常見踩坑經(jīng)驗案例一async def端點中的time.sleep()卡死事件循環(huán)現(xiàn)象100 并發(fā)請求只有一個請求在執(zhí)行其余 99 個排隊——QPS 只有 0.5。根因開發(fā)者在async def函數(shù)中調(diào)了time.sleep(2)事件循環(huán)被阻塞 2 秒。解決await asyncio.sleep(2)或改用def端點讓線程池處理。案例二asyncio.gather中一個任務(wù)掛起導(dǎo)致所有任務(wù)超時現(xiàn)象3 個 API 并發(fā)調(diào)用其中一個超時 30s其余兩個 200ms 就返回了一直被攔住。根因gather默認等待所有任務(wù)完成才返回。解決為每個任務(wù)單獨設(shè)置 timeout —asyncio.wait_for(task, timeout5)或使用asyncio.as_completed()先返回先處理。案例三httpx.AsyncClient提前關(guān)閉現(xiàn)象服務(wù)啟動正常運行幾分鐘后所有外部 API 調(diào)用報RuntimeError: Event loop is closed。根因在 Lifespan 中創(chuàng)建了AsyncClient但在某次異常中沒有正確關(guān)閉。下次請求時復(fù)用了一個半關(guān)閉的 client。解決在app的 lifespan 事件中管理 client 的創(chuàng)建和關(guān)閉或每次請求創(chuàng)建新的AsyncClient性能略低但更安全。思考題初級修改報價服務(wù)新增一個超時兜底模式——如果某個 API 在 1 秒內(nèi)未響應(yīng)使用緩存中的上一次數(shù)據(jù)作為兜底stale-while-revalidate 策略。進階如何使用asyncio.TaskGroupPython 3.11替代asyncio.gatherTaskGroup相比gather的優(yōu)勢是什么提示結(jié)構(gòu)化并發(fā)。答案提示第 1 題使用asyncio.wait_for(task, timeout1)配合緩存。第 2 題TaskGroup是 Python 的結(jié)構(gòu)化并發(fā)原語——如果組內(nèi)任一任務(wù)拋異常所有子任務(wù)自動取消不會出現(xiàn)孤兒協(xié)程。第 37 章深入事件循環(huán)診斷與性能極限。延伸閱讀與資源NumPy 從入門到生產(chǎn)落地全鏈路實戰(zhàn)指南科學(xué)計算/向量化Redis 8 實戰(zhàn)精講從 CRUD 到源碼構(gòu)建高可用緩存系統(tǒng)Redis 實戰(zhàn)修煉與原理進階Python 3實戰(zhàn)精進從腳本到高并發(fā)訂單引擎python入門Rquests從菜鳥腳本到企業(yè)級SDK的網(wǎng)絡(luò)實戰(zhàn)圣經(jīng)Milvus向量數(shù)據(jù)庫實戰(zhàn)修煉從 0 到 1精通向量檢索與生產(chǎn)落地MongoDB 實戰(zhàn)進階與內(nèi)核修煉后端工程師的 AI 轉(zhuǎn)型第一課Ollama 與私有化大模型實戰(zhàn)10倍開發(fā)者的 Dify 魔法書從零構(gòu)建全棧 AI 應(yīng)用后端工程師轉(zhuǎn)型AI第一課-Ollama 與私有化大模型實戰(zhàn)大型語言模型(LLM) vLLM 高性能推理落地實戰(zhàn)Agent開發(fā)之LlamaIndex 實戰(zhàn)修煉與源碼進階大語言模型Transformers 實戰(zhàn)修煉與源碼剖析