做 AI 應用最難受的時刻之一,就是你明明調的是流式接口,結果用戶那邊看到的還是一個轉圈圈,過了四五秒才“嘩啦”一下全出出來。更難受的是,后端根本沒報錯,線程池卻悄悄死了,接口一個接一個超時,進程沒掛,但整個服務已經“假死”。這種問題我見過太多次,也親手排查過太多回,根子往往就一句話:線程被堵死了。這篇文章圍繞“AI 流式輸出”和“異步調度”這兩件事展開,核心始終是標題里那五個字——別堵死線程。我會把流式接口的底層協議、線程池被堵死的完整連鎖過程、以及從同步到異步的改造實操都講透,包括很多不踩坑你根本不會留意的細節(jié)。不管你做的是 Java 還是 Python 后端,是業(yè)務接口還是網關層,都能直接拿去參考。1. 流式輸出的本質:一次“打字機”式的技術改造1.1 從“等結果”到“看結果”的體驗飛躍先聊為什么大家非要流式。用過 ChatGPT、Claude 這類產品的人都有感覺:AI 生成回答不是瞬間完成的,一個三四百字的回答,模型可能要“想”5 到 20 秒。如果走傳統(tǒng)的同步接口,這 5 到 20 秒里用戶只能盯著一張空白頁面,什么反饋都沒有。產品數據也說明問題:超過 3 秒沒有反饋,用戶流失率會直線上升。我自己做過的 AI 問答項目里,首字延遲從 8 秒壓到 1 秒以內,用戶滿意度提升非常明顯,這就是流式的價值。流式輸出(Streaming Output)解決的就是這個體驗問題。它把模型一次性的長響應,拆成一個個 token(你可以理解為詞或子詞),邊生成邊推送。用戶在界面上看到的效果是文字一個字一個字往外蹦,像有人在打字。很多 AI 產品的“打字機效果”就是這么來的。但這里有個容易忽略的坑:流式只是改變了“數據怎么返回”,并沒有改變“模型算多久”。從調用方視角看,總耗時是一樣的,甚至因為要持續(xù)維持連接,占用的資源更多。也就是說,流式把壓力從“用戶等待”轉移到了“后端長連接維護”上,這才是后面一堆線程問題的根源。很多人以為用了流式就萬事大吉,結果發(fā)現服務端更吃緊了,原因就在這里。1.2 流式接口底層到底發(fā)生了什么拿目前用得最多的 OpenAI 兼容接口來說,流式響應走的是 SSE(Server-Sent Events)。SSE 本質就是 HTTP 長連接,服務端按text/event-stream格式,一條一條往客戶端推數據。每條事件以data:開頭,以空行結束。AI 流式接口的事件體通常是 JSON,里面帶choices[0].delta.content字段,收到的就是一個增量片段。舉個例子,真實抓包看到的流大概是這種:data: {choices:[{delta:{content:你},index:0}]} data: {choices:[{delta:{content:好},index:0}]} data: {choices:[{delta:{content:,},index:0}]} ... data: [DONE]注意幾個細節(jié):第一,這串內容極其碎片化,一個詞可能拆成好幾次推送,甚至一個字分多個事件;第二,HTTP 層用的是 chunked transfer encoding,也就是 TCP 連接上數據是一塊一塊流過來的,不是一次性寫完;第三,如果客戶端沒及時讀數據,服務端緩沖區(qū)滿了會反壓(TCP backpressure),這時候模型那邊也會等。這就帶出一個關鍵認知:fetch、HttpClient、requests 這些庫都能處理流式響應,關鍵看你怎么調。大多數 HTTP 客戶端默認“等全部響應體到達再返回”,即使你設置了streamtrue,底層如果沒配合逐行讀取,一樣會攢到全部數據才 yield。我見過不少“明明用了流式接口,前端還是等全量出來”的案例,問題就出在這一層。所以別以為配個 stream 參數就完事了,讀響應的方式才是決定成敗的細節(jié)。2. 為什么你的線程會被“堵死”2.1 同步等待是萬惡之源在線程模型下,一次同步調用意味著“當前線程發(fā)起請求后,進入等待狀態(tài),直到響應完全返回”。AI 接口的響應時間有多長?慢的時候 60 秒甚至更久。這段時間里,那條線程做不了任何事,就在那等著。你可以把線程想象成餐廳的服務員,如果每個服務員都站在一張桌子旁邊等客人慢慢吃完,這餐廳就沒法接待別的客人了。假設你的服務用的是 Tomcat 默認配置,工作線程上限 200。某個活動上線后,10 個用戶同時發(fā)起 AI 對話請求,每個請求都要等 20 秒。這 10 條線程就被占住了,相當于 10/200 的容量沒了。用戶一多,比如 200 個并發(fā)請求,所有工作線程全部被占滿,后面的請求直接排隊,排不上就超時。你在監(jiān)控里看到的往往就是“TPS 掉到 0 但 CPU 只有 5%”,因為線程全部在等網絡,根本沒在算。Java 里這種問題最容易藏。很多同學用 RestTemplate、OkHttp 同步調用大模型接口,一個execute()下去,當前線程就進去了。代碼看起來只有一行,非常清爽,但坑就在下面深著呢。用Thread.currentThread().getName()把執(zhí)行線程名打出來看看,你會發(fā)現同樣一條 Tomcat 工作線程,從請求進來一直陪到 AI 返回,中間啥也沒干。2.2 線程池耗盡:一次連鎖反應的現場還原我復盤過好幾次線上假死,套路都差不多。這里還原一遍,方便你對號入座。第一步,業(yè)務高峰期,多個請求同時調 AI 流式接口,每個占住一條 Tomcat 工作線程,等待模型返回。第二步,Tomcat 的 200 條線程被占滿,后續(xù)所有請求(包括不涉及 AI 的普通接口)全部排隊。第三步,等待隊列塞滿,新請求觸發(fā)拒絕策略,默認的 AbortPolicy 直接拋RejectedExecutionException,調用方 500。第四步,前端收到 500 就自動重試,重試又涌進來,線程池進一步被沖垮。更糟的是,如果這些請求是從某個 Spring 異步線程池發(fā)出去的,情況會復雜一層:業(yè)務線程池也在等,HTTP 連接池也在等,連接池超時后又波及到別的服務。這就是典型的“一線程堵死,全線崩潰”。問題不在 AI,而在你把一條寶貴的線程押在了不可控的等待時長上。線程池這東西,看起來只是“線程的集合”,實際上它是整個系統(tǒng)的排隊機制,一旦它被打滿,所有依賴它的調用都會跟著遭殃。2.3 線程池參數怎么配:算給你看如果一定要用線程池(有些場景確實只能用線程池),參數不能拍腦袋。核心公式是:對于 IO 密集型任務:線程數 CPU 核數 × (1 等待時間 / 計算時間)這里的“計算時間”對 AI 流式場景來說,基本就是 JSON 解析和業(yè)務處理的耗時,通常在幾毫秒到幾十毫秒;而“等待時間”是模型響應時間,15 秒起步。假設 1 次請求等待 15 秒,計算 10 毫秒,比值是 1500。一個 8 核機器按公式算出來理論線程數是 8 × (1 1500) 12008 條線程。這個數字顯然不能直接照搬,因為線程是又貴又不可能無限開的資源,每條線程默認 1MB ?？臻g,12000 條線程光棧就 12GB,內存直接爆。所以實戰(zhàn)中要反過來算:先看你能給多少內存,再定線程數上限。8GB 堆、單線程默認棧 1MB 的設備,線程數壓到 500 以內比較穩(wěn)妥。再疊加下游 AI 服務的并發(fā)限制(比如每分鐘 60 次)和 HTTP 連接池上限,真正合理的線程數往往是 100 到 300 之間。配置線程池的順序應該是先算內存上限,再算下游限流,最后才是 CPU 公式。這個順序很多人反著來,導致各種詭異問題。3. 異步調度的正確打開方式3.1 Java 的異步進化:從 Future 到 CompletableFuture堵死線程的解法不是不創(chuàng)建線程,而是別讓線程“干等”。Java 里最基礎的異步是Future,允許你提交任務后不阻塞,后面再get()。但Future.get()本身就是同步阻塞,如果你在業(yè)務代碼里無腦調用,問題照舊。真正的異步模型應該是:提交任務、立即返回、回調通知,全程不等待。正確姿勢是用CompletableFuture。它支持回調、組合,最關鍵的是supplyAsync()可以指定獨立的線程池,把 AI 調用從 Tomcat 請求線程挪到專門的工作線程上,請求線程立刻釋放。比如:CompletableFutureString future CompletableFuture.supplyAsync( () - aiClient.complete(prompt), // 這個階段在 aiExecutor 中執(zhí)行 aiExecutor ); future.thenAccept(result - log.info(AI 返回完成,當前線程:{}, Thread.currentThread().getName()));重點是thenAccept之后的邏輯不會占用 Tomcat 工作線程。future本身可以被繼續(xù)傳遞,也可以配合CompletableFuture.allOf()等所有 AI 子任務都完成后統(tǒng)一處理。如果你要“等全部完成再匯總”,別用循環(huán)future.get(),那是變相把異步變回同步;應該用 allOf:ListCompletableFutureString futures prompts.stream() .map(p - CompletableFuture.supplyAsync(() - aiClient.complete(p), aiExecutor)) .collect(Collectors.toList()); CompletableFutureVoid allDone CompletableFuture.allOf( futures.toArray(new CompletableFuture[0])); allDone.join(); // 這一行只在自己確實需要同步等待時才用這里有個非常容易被忽略的點:如果回調也在aiExecutor里執(zhí)行,回調里的代碼也會有同樣的問題。所以我習慣給不同階段分開線程池,或者干脆用事件驅動的方式,不讓回調做重計算。另外注意,CompletableFuture默認用的是 ForkJoinPool,里面的線程數等于 CPU 核數,你如果把 AI 這種幾十秒的 IO 等待任務往里塞,這個池子瞬間被占滿,項目里其它所有用 CompletableFuture 的地方都會跟著卡住。給 AI 調用單獨建線程池是必須的,不是可選項。3.2 Python 那邊的玩法:協程與線程池的搭配Python 走的是另一條路:GIL 決定了純線程在 CPU 密集型任務上沒法并行,但對 IO 等待反而友好。因為等待期間線程讓出 GIL,別的線程能跑。問題是,如果你用requests這種同步庫,即使底層是 IO 等待,線程也會被占住,只是 GIL 讓出了 CPU 而已。從線程占用角度看,占住就是占住,和 GIL 沒關系。Python 的推薦做法是異步框架加異步 HTTP 客戶端。FastAPI httpx 的AsyncClient.stream()是個典型組合,流式讀取 SSE 時不會阻塞事件循環(huán):from fastapi import FastAPI from fastapi.responses import StreamingResponse import httpx app FastAPI() async def sse_proxy(prompt: str): payload { model: gpt-4o-mini, messages: [{role: user, content: prompt}], stream: True } async with httpx.AsyncClient(timeouthttpx.Timeout(60.0, read60.0)) as client: async with client.stream( POST, https://api.example.com/v1/chat/completions, jsonpayload ) as resp: async for line in resp.aiter_lines(): if line.startswith(data:): yield line \n\n app.get(/chat) async def chat(prompt: str): return StreamingResponse(sse_proxy(prompt), media_typetext/event-stream)注意timeout有多個參數:connect 超時和 read 超時是兩回事。AI 流式接口的 read 超時不能設太短,因為模型可能“想”很久不吐詞,但你又不能完全沒有超時。我的經驗是把 connect 超時設在 3 到 5 秒,read 超時設在 60 到 120 秒,再把一個“首個 token 超時”的邏輯做在應用層(比如 30 秒內沒收到第一個 data 就斷掉)。Python 還有一個常見的坑:線程嵌套線程。有些同學為了“更并發(fā)”,在異步代碼里偷偷調threading.Thread,或者用asyncio.to_thread包一層同步庫,結果主事件循環(huán)沒堵,但子線程全堵在下游接口上,線程數越開越多。Python 線程切換本身也有開銷,特別是在threading混合asyncio的場景,很容易出現上下文切換暴漲。原則是:一條鏈路盡量只用一個并發(fā)模型,要么全異步,要么全線程池,別混著來。順帶提一句,Python 3.13 開始有了 free-threading 模式,可以真正并行跑線程,但對 IO 等待場景幫助不大,因為等待本來就會讓出 GIL。不要因為聽說 GIL 要沒了,就把同步阻塞代碼到處撒線程,線程創(chuàng)建和切換的開銷依舊在那。3.3 響應式編程是不是唯一出路很多人一聽異步就說要用 WebFlux、Reactor。響應式確實是解決線程阻塞的好辦法,在 Spring 生態(tài)里,FluxString配合 SSE 可以直接把流式響應串成響應式流:GetMapping(value /chat, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxString chat(String prompt) { return Flux.create(sink - { aiClient.completeStream(prompt, sink::next, sink::complete); }).subscribeOn(Schedulers.boundedElastic()); }但響應式不是銀彈。它最大的問題是調試困難,調用棧是斷的,異常處理邏輯跟同步代碼完全不同,團隊沒有響應式經驗的話,上線后維護成本很高。我做過的項目里,真正的選擇依據是:如果只是“代理大模型流式響應給前端”,那我更傾向于用最簡單的方式——前端能直接連的就直接連,不能直接連的用 Spring 的SseEmitter或者 Servlet 3.1 的異步支持,別一上來就上 WebFlux。如果你在做一個高并發(fā)網關,要支撐成千上萬的 SSE 長連接,那響應式才劃算。技術選型不是越新越好,是越合適越好。Redis 的線程 IO 模型就很有啟發(fā)性:主線程用單線程事件循環(huán)處理命令,網絡 IO 由多線程處理,真正體現了“不等待”的哲學。做 AI 網關的時候,可以參考這種思路,把網絡讀寫和業(yè)務處理解耦,而不是靠堆線程硬扛。4. 實操改造:一個完整的流式接口從堵到順4.1 改造前后的代碼對比紙上談兵沒意思。我拿一個真實的小項目改造過程來演示,原先是 Spring Boot RestTemplate 同步調 AI,改成 Spring MVC SseEmitter 獨立線程池。這個項目是個簡單的 AI 聊天問答服務,前端只有一句話輸入框和一塊顯示區(qū)域,但上線后并發(fā)一高就頻繁超時,這才被迫做了改造。改造前的問題代碼大概長這樣:GetMapping(/chat) public String chat(String prompt) { // 這一步會阻塞 Tomcat 工作線程最長 60 秒 ChatResponse resp restTemplate.postForObject( https://api.example.com/v1/chat/completions, buildRequest(prompt, false), ChatResponse.class ); return resp.getContent(); }這段代碼在低并發(fā)下沒問題,一壓測就原形畢露。改造后:GetMapping(value /chat, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter chat(String prompt) { SseEmitter emitter new SseEmitter(120_000L); // 2 分鐘超時 aiExecutor.execute(() - { try { aiClient.completeStream(prompt, chunk - { emitter.send(SseEmitter.event().data(chunk)); }); emitter.complete(); } catch (Exception e) { emitter.completeWithError(e); } }); return emitter; }關鍵改動只有兩處:第一,接口不再返回字符串,而是返回SseEmitter,Spring 會拿它做 SSE 推送;第二,真正的 AI 調用丟給aiExecutor執(zhí)行,Tomcat 請求線程立刻返回。Tomcat 工作線程的占用時間從“最長 60 秒”降到了“幾毫秒”。這里有個細節(jié):SseEmitter 超時時間一定要大于你的 read 超時。如果 AI 最慢 60 秒返回,你把 emitter 超時設成 30 秒,客戶端就會莫名斷連,看起來像 AI 吐了一半被砍了。同樣,前端 fetch 也要把readTimeout調大,否則瀏覽器 20 秒沒收到數據就報網絡錯誤。這種問題最坑,因為后端日志顯示一切正常,斷連發(fā)生在瀏覽器那一層。4.2 異步網關層如何把流“接力”給前端如果你的架構里還有一層網關(比如自研的 API 網關,或者公司的 BFF 層),流式響應在每一層都得“邊收邊轉”。最容易翻車的地方是網關層用同步 HTTP 客戶端去請求后端,要么全量等完再往前端吐,要么直接超時。正確的接力邏輯是:網關收到后端傳來的 SSE 流,逐條解析data:行,然后立即把同一段數據寫到下游響應。也就是從“后端 → 網關 → 前端”這條鏈路上,每一跳都是流式的。如果用 Java 寫網關,HttpURLConnection和舊 HttpClient 都不支持流式讀響應體,建議直接用java.net.http.HttpClient(Java 11 起內置)的BodyHandlers.ofInputStream(),或者用 OkHttp 的onResponse回調里逐行讀。另外,網關層最好別做“攢批”:每收到幾個 chunk 才往客戶端推一次,來降低推送頻率。這種做法表面上看似省 IO,實際上會在鏈路上引入額外延遲,而且一旦中間斷開,損失的已經不是一條消息,是一整批。我的實踐是:來什么推什么,保持流數據的透傳。壓縮可以開,但要用流式壓縮(如 gzip 的連續(xù)模式),不能等整個流結束才壓縮,否則就失去了流式的意義。4.3 承載量評估:改造后能扛多少并發(fā)讀者最關心的肯定是“改完到底能扛多少”。我拿一個真實壓測數據來說明。某項目原先用同步方式,8 核 16G 的機器,Tomcat 默認 200 線程,壓測 50 并發(fā),每個請求平均耗時 20 秒,實際吞吐不到 10 QPS,線程池直接打滿。改造后用 200 線程的獨立aiExecutor處理 AI 調用,Tomcat 工作線程只負責接收和返回 emitter。壓測結果:同一臺機器,200 并發(fā),每個請求體驗不受影響,吞吐接近 150 QPS,而且 Tomcat 工作線程占用率不到 60%。瓶頸轉移到下游 AI 服務的限流上。這個提升的核心原因是:Tomcat 工作線程不再閑置等待,線程復用率大幅提高。說白了,200 條線程在同步模型下頂多同時服務 200 個“正在等待中”的請求;在異步模型下,200 條線程可以服務幾千個“已受理、等待中”的請求,因為每個請求真正占用線程的時間只有幾毫秒。這就是“別堵死線程”的量化意義。5. 常見問題與排查技巧實錄5.1 線程死鎖、池耗盡、連接泄漏三巨頭AI 流式場景里,線上問題逃不出這三類,我一個個說。線程死鎖:典型的現場是 A 線程等待 B 線程的結果,而 B 線程也在等 A 線程的資源。在 AI 調用場景,死鎖經常藏在“AI 超時重試”的代碼里。比如線程池被打滿后,重試任務又被提交到同一個已滿線程池,新任務永遠等不到執(zhí)行,形成變相死鎖。另一個常見情況是CompletableFuture的默認 ForkJoinPool 被 AI 阻塞任務占滿,導致所有使用它的異步任務集體卡住。所以給 AI 調用單獨建線程池是必須的,還要給get()和join()設置超時,不能讓等待無限繼續(xù)。線程池耗盡:這個上文分析過。排查時先看線程數。Linux 下top -H -p pid能看到線程快照,如果發(fā)現有幾百條線程狀態(tài)都是D(不可中斷睡眠)或長時間S(睡眠),基本可以斷定它們在等網絡 IO。拿jstack導出線程轉儲,grep一下 WAITING 狀態(tài)線程的調用棧,十有八九能看到指向 HTTP 客戶端read方法的那一行。連接泄漏:很多 AI HTTP 客戶端默認不會無限復用連接。如果你在每次請求里 new 一個 OkHttpClient,或者不關閉響應體,連接池會被慢慢耗盡。更隱蔽的是連接池的maxConnections設置太小,比如默認 20,而你開了 100 個并發(fā)流,剩下 80 個請求全部在等連接,表現起來和線程池耗盡一模一樣。排查這種問題,用lsof -p pid | grep TCP看看連接數,再檢查 HttpClient 的池參數。5.2 現場排查:拿到線程轉儲怎么讀jstack 是 Java 排查線程問題的核心工具,但很多人拿到輸出后一頭霧水。我說一個快速的閱讀方法。先看狀態(tài)分布,統(tǒng)計 RUNNABLE、WAITING、BLOCKED、TIMED_WAITING 各自的數量。正常服務 RUNNABLE 應該占大頭;如果 WAITING 占絕大多數,而且線程名集中在http-nio-8080-exec-*,那基本可以斷定是工作線程池被打滿。找到一條可疑線程,看它的調用棧底部。比如看到:at java.base/sun.nio.ch.NioSocketImpl.read(NioSocketImpl.java:139) at org.apache.hc.client5.http.impl.io.ChunkedInputStream.read(...)這說明線程卡在“讀分塊響應”上,也就是正在等上游 AI 吐數據。結合線程數統(tǒng)計,如果大量線程都停在這里,問題定位就完成了:下游 AI 響應慢,導致工作線程被占滿。Windows 下有人會用 Process Explorer 看線程,甚至可以凍結(freeze)某個線程來模擬卡死現象。凍結線程對本地調試重啟時的條件競爭很有用,但千萬別在生產環(huán)境這么干,凍結的是進程里某個線程的調度,會讓依賴它的請求直接假死,而且 Process Explorer 并不會告訴你為什么會卡,它只是給你一個“手動制造故障”的工具。線上排查還是靠線程轉儲和監(jiān)控日志。5.3 排查工具與避坑清單最后整理一份實用的避坑清單,這些都是我踩過或者幫別人擦過屁股的:第一,所有 HTTP 超時都要顯式設置。連接超時 3 到 5 秒,讀取超時 60 到 120 秒,并且最終響應(整體)超時要比這些更大。第二,給 AI 調用單獨配置線程池,不要在Async默認執(zhí)行器、ForkJoinPool 或 Tomcat 工作線程里跑模型調用。第三,流式接口一定設置媒體類型為text/event-stream,并且關閉響應緩沖。Spring 里如果忘了設produces,或者 Nginx 層開了proxy_buffering on,前端可能等全量結束才看到數據,等于白做流式。第四,線程池要用有界隊列和合理的拒絕策略。隊列選ArrayBlockingQueue,別用無界LinkedBlockingQueue,否則任務無限排隊,內存先爆。拒絕策略要么CallerRunsPolicy,要么自定義告警,不要讓拒絕默默發(fā)生。第五,守護線程要慎用。如果你用ExecutorService創(chuàng)建了守護線程處理 AI 回調,當主線程結束、JVM 里只剩守護線程時,進程會直接退出,長連接全部斷開。自定義線程工廠時把線程設為非守護。第六,AtomicInteger在計數線程池任務數、請求數時沒問題,但別拿它來做線程安全的業(yè)務計數,因為它只保證單個變量的原子性,不保證復合操作的原子性。真要統(tǒng)計 A 線程池里的活躍任務數,還是老老實實等ThreadPoolExecutor的getActiveCount,或者自己加鎖保護。第七,在 Python 異步代碼里混用同步阻塞庫,是線程切換開銷的隱形殺手。每個asyncio.to_thread都會創(chuàng)建線程,如果循環(huán)調用,線程數會失控。能用異步庫(httpx、aiohttp)就盡量別用同步庫。第八,針對“線程切換時會泄漏嗎”這個問題,答案是:切換本身不會泄漏,但線程局部變量(ThreadLocal)在線程復用池里會。池里的線程不會因為任務結束就被回收,ThreadLocal里的數據如果沒有清理,下一個任務復用這條線程時能讀到上一個任務的值。AI 請求里如果往 ThreadLocal 塞了用戶上下文,記得在 finally 里 remove。第九,開始排查之前先確認你看到了真相:連接池耗盡、線程池耗盡、下游慢、上游超時,這四類問題的表象都是“請求變慢、開始超時”,但解法完全不同。拿線程轉儲、連接數、GC 日志三個數據源交叉驗證,別靠猜。6. 寫在最后的幾句體會其實不存在“絕對正確”的架構,只要記住一點:在 AI 這種動輒幾十秒的 IO 等待面前,任何讓線程干等的設計都是犯罪。線程是拿來干活的,不是拿來排隊的。你真正需要的是把“發(fā)起調用”和“等待結果”拆開,用事件或者回調把結果送回來,而不是讓一條線程從頭盯到尾。我個人實操中的體會是:一開始別追求什么高大上的響應式,先把同步調用改成異步線程池 回調,把超時和連接池配好,能解決 80% 的問題。剩下 20% 的高并發(fā)場景,再考慮 WebFlux 或者獨立的 SSE 網關。技術的選擇要跟著你的流量走,而不是跟著技術熱度走。希望這篇文章能幫你少踩幾個坑,下次線上再“假死”的時候,至少能知道該從哪里下手。