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