1. 從一次“打字機卡頓”說起流式傳輸到底解決了什么很多人第一次接觸流式傳輸是在做 AI 對話界面的時候。用戶點下發(fā)送按鈕界面上轉圈圈等了七八秒整段回答“啪”地一下全冒出來。體驗上就像打電話時對方一直不說話等你想掛斷的時候他突然把整段話一口氣說完。流式傳輸要解決的就是把這個“憋大招”的過程拆開讓內容像打字機一樣一個字一個字往外蹦。我最早做類似功能時腦子里只有一個樸素的想法后端生成一段文字前端顯示一段文字中間用普通的 HTTP 請求不就行了實測下來問題很明顯——普通 HTTP 是“請求-響應”模型服務端必須把完整響應體準備好才能一次性發(fā)給客戶端。大模型生成一段 500 字的回答可能要十幾秒這十幾秒里前端什么都拿不到只能干等。用戶不知道后臺是在正常工作還是已經掛了體驗非常糟糕。流式傳輸的核心思路是把一份完整的數據切成很多小塊服務端生成一塊就發(fā)一塊客戶端收到一塊就渲染一塊。這樣用戶看到的是內容在持續(xù)增長心理上會覺得“系統(tǒng)在干活”等待焦慮大幅降低。這個思路并不新鮮視頻網站早就在用流媒體你拖動進度條時視頻不是全部下載完才播放而是邊下邊播。AI 對話場景只是把“視頻幀”換成了“文本片段”。那為什么大家總把流式傳輸和 SSE 協(xié)議綁在一起講因為 SSEServer-Sent Events服務器推送事件是瀏覽器原生支持的一種“服務端持續(xù)向客戶端推送文本”的機制它天然適合“服務端生成、客戶端展示”這種單向數據流。相比 WebSocket 的雙向通信SSE 更輕、更簡單用普通 HTTP 連接就能跑不需要額外協(xié)議升級。對于大模型回答這種“我問一句、你答一長段”的場景SSE 的匹配度非常高。這篇文章我會從實際項目出發(fā)把流式傳輸的原理、SSE 協(xié)議的細節(jié)、前后端怎么配合、Abort 中斷怎么處理、以及我踩過的那些坑一層一層拆開講。如果你正在做 AI 對話、實時日志、進度推送這類功能或者只是單純想搞明白“為什么別人的回答能一個字一個字往外蹦”這篇內容應該能幫你省下不少查資料的時間。2. SSE 協(xié)議的真實面目它不是什么黑科技就是一段有格式的文本2.1 SSE 的報文格式四個字段撐起整個協(xié)議很多人覺得 SSE 很神秘其實它簡單到有點“簡陋”。SSE 的本質是服務端保持一個 HTTP 連接不關閉然后按照固定格式往這個連接里寫文本。瀏覽器收到這些文本后按照同樣的格式解析觸發(fā)對應的事件。整個協(xié)議的核心字段只有四個字段作用是否必需data消息內容可以多行是event自定義事件類型默認是 message否id消息編號用于斷線重連時定位否retry重連等待時間毫秒否一條典型的 SSE 消息長這樣event: message id: 1 data: {content: 你} data: {content: 好}注意幾個細節(jié)每個字段后面跟一個冒號和一個空格然后才是值一條消息以兩個換行符結束data可以出現(xiàn)多次瀏覽器會把它們用換行符拼起來。我第一次手寫 SSE 服務端時就是因為少寫了一個換行前端死活收不到消息排查了半小時才發(fā)現(xiàn)是格式問題。2.2 和 WebSocket 的取舍為什么 AI 對話場景更偏愛 SSE剛接觸這兩個技術的人經常會問既然 WebSocket 能雙向通信看起來更強大為什么 AI 對話場景大多用 SSE我自己的判斷邏輯是這樣的通信方向AI 對話是典型的“客戶端發(fā)一次請求服務端持續(xù)返回”。請求只有一次返回有很多次。SSE 的單向推送剛好匹配WebSocket 的雙向能力在這里是浪費的。實現(xiàn)成本SSE 用普通 HTTP 就能跑服務端就是往響應流里寫字符串客戶端用EventSource幾行代碼就能接。WebSocket 需要協(xié)議升級、心跳?；?、重連邏輯復雜度高一個量級?；A設施兼容SSE 走的是標準 HTTP現(xiàn)有的網關、負載均衡、日志系統(tǒng)基本都能直接處理。WebSocket 的升級握手在某些代理環(huán)境下容易被攔截排查起來很頭疼。自動重連EventSource內置了斷線重連機制服務端可以通過retry字段控制重連間隔。WebSocket 的重連得自己寫。當然 SSE 也有明顯短板它只能服務端推客戶端客戶端要發(fā)消息得另開一個 HTTP 請求它傳輸的是文本二進制數據需要額外編碼瀏覽器對同一域名的 SSE 連接數有限制HTTP/1.1 下通常是 6 個。所以如果你的場景是聊天室、協(xié)同編輯這種雙向高頻通信WebSocket 更合適。但如果是“請求一次、持續(xù)接收”SSE 是更省事的選擇。2.3 瀏覽器端的 EventSource好用但有邊界瀏覽器原生提供了EventSource對象來接收 SSE 流用法簡單到離譜const es new EventSource(/api/chat/stream); es.onmessage (event) { console.log(收到消息:, event.data); }; es.onerror (err) { console.error(連接出錯:, err); };但EventSource有幾個讓人難受的限制我在項目里都遇到過第一它只支持 GET 請求。這意味著你沒法在請求體里放復雜的參數只能把參數拼在 URL 上。對于 AI 對話這種需要傳對話歷史、模型參數、系統(tǒng)提示詞的場景URL 長度很容易超限而且把敏感信息放在 URL 里也不安全。第二它不能自定義請求頭。你沒法加Authorization頭做鑒權只能靠 Cookie 或者 URL 參數傳 token。第三它不能中斷請求。EventSource只有close()方法關閉連接但沒有“主動取消”的語義服務端可能還在繼續(xù)生成資源就浪費了。正因為這些限制現(xiàn)在很多 AI 對話項目并不直接用EventSource而是用fetch配合ReadableStream手動解析 SSE 流。這樣既能用 POST 傳參、自定義請求頭又能通過AbortController隨時中斷。代價是你得自己寫解析邏輯不能白嫖瀏覽器的自動重連。3. 用 fetch 手動接管 SSE把控制權拿回自己手里3.1 為什么放棄 EventSource 轉向 fetch前面說了EventSource的三個硬傷其中“不能中斷”和“不能 POST”對 AI 對話來說是致命的。用戶點了發(fā)送等了三秒覺得不對想取消EventSource做不到對話歷史有十幾輪全塞 URL 里也不現(xiàn)實。所以我在實際項目里基本都用fetch來手動處理 SSE 流。fetch的優(yōu)勢在于它返回的response.body是一個ReadableStream你可以一塊一塊地讀讀到什么就處理什么。同時fetch支持AbortController想中斷隨時中斷。請求方法、請求頭、請求體全都自由。缺點就是 SSE 的解析得自己寫但這段邏輯并不復雜封裝一次就能到處用。3.2 手動解析 SSE 流的關鍵代碼先看服務端返回的響應頭必須設置正確否則瀏覽器可能會緩沖整個響應Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive X-Accel-Buffering: noX-Accel-Buffering: no這個頭是給 Nginx 看的告訴它不要緩沖這個響應。我踩過一次坑本地開發(fā)一切正常部署到有 Nginx 的服務器后流式效果消失了所有內容一次性冒出來。排查半天才發(fā)現(xiàn)是 Nginx 默認開啟了代理緩沖把 SSE 流攢著一起發(fā)了。加上這個頭或者在 Nginx 配置里關掉proxy_buffering問題就解決了?？蛻舳私馕龅暮诵倪壿嫶蟾攀沁@樣async function streamChat(url, body, onChunk, signal) { const response await fetch(url, { method: POST, headers: { Content-Type: application/json, Accept: text/event-stream, }, body: JSON.stringify(body), signal, }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); // 按雙換行切分消息 const parts buffer.split(\n\n); buffer parts.pop(); // 最后一段可能不完整留到下次 for (const part of parts) { const lines part.split(\n); for (const line of lines) { if (line.startsWith(data: )) { const data line.slice(6); if (data [DONE]) return; onChunk(data); } } } } }這段代碼有幾個關鍵點值得展開說。decoder.decode(value, { stream: true })里的stream: true很重要它保證多字節(jié)字符比如中文被正確拼接。如果不加這個參數一個中文字符被切成兩個字節(jié)分別解碼就會出現(xiàn)亂碼。buffer的作用是處理“半條消息”——網絡傳輸不會按照你的消息邊界來切可能一條消息只到了一半你得把它留到下一輪再拼。3.3 中斷請求AbortController 的正確用法AbortController是配合fetch實現(xiàn)中斷的標準方案。創(chuàng)建一個 controller把它的signal傳給fetch需要中斷時調用controller.abort()const controller new AbortController(); // 發(fā)起請求 streamChat(/api/chat, { message: 你好 }, onChunk, controller.signal); // 用戶點擊停止按鈕 stopButton.onclick () { controller.abort(); };調用abort()后fetch的 promise 會拋出一個AbortErrorreader.read()也會立即結束。你需要在代碼里捕獲這個錯誤避免它冒泡到全局try { await streamChat(...); } catch (err) { if (err.name AbortError) { console.log(用戶主動中斷了請求); } else { console.error(請求出錯:, err); } }這里有個容易忽略的點客戶端中斷了服務端不一定知道。fetch斷開連接后服務端往響應流里寫數據會失敗但服務端代碼如果沒做檢查可能還在傻傻地調用大模型接口白白消耗 token。所以服務端也要監(jiān)聽連接關閉事件及時停止生成。在 Node.js 里可以監(jiān)聽req.on(close)在 Python 的 FastAPI 里可以監(jiān)聽request.is_disconnected()。4. 服務端怎么把流“推”出去不同技術棧的落地方式4.1 Node.js 原生寫法res.write 就夠了Node.js 的http模塊天然支持流式響應核心就是設置好響應頭然后不斷調用res.write()app.post(/api/chat/stream, async (req, res) { res.setHeader(Content-Type, text/event-stream); res.setHeader(Cache-Control, no-cache); res.setHeader(Connection, keep-alive); res.setHeader(X-Accel-Buffering, no); // 監(jiān)聽客戶端斷開 let aborted false; req.on(close, () { aborted true; }); const stream await callLLM(req.body.message); for await (const chunk of stream) { if (aborted) break; res.write(data: ${JSON.stringify({ content: chunk })}\n\n); } res.write(data: [DONE]\n\n); res.end(); });注意res.write()的格式data:前綴加上內容然后兩個換行符。少一個換行前端就解析不出來。[DONE]是一個約定俗成的結束標記OpenAI 的接口就是這么干的前端收到它就停止讀取。4.2 Python FastAPIStreamingResponse 的坑FastAPI 提供了StreamingResponse來簡化流式輸出from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app FastAPI() async def event_generator(request: Request, message: str): async for chunk in call_llm(message): if await request.is_disconnected(): break yield fdata: {json.dumps({content: chunk})}\n\n yield data: [DONE]\n\n app.post(/api/chat/stream) async def chat_stream(request: Request): body await request.json() return StreamingResponse( event_generator(request, body[message]), media_typetext/event-stream, headers{ Cache-Control: no-cache, X-Accel-Buffering: no, }, )這里有個坑我印象很深FastAPI 的StreamingResponse默認會經過一些中間件如果中間件對響應做了緩沖流式效果就沒了。另外request.is_disconnected()的檢測不是實時的它依賴于底層連接狀態(tài)有時候客戶端已經斷了服務端還要再寫一兩次才發(fā)現(xiàn)。所以更穩(wěn)妥的做法是結合超時機制別讓生成任務無限跑下去。4.3 大模型接口的流式返回OpenAI 格式的解析現(xiàn)在主流大模型接口都支持流式返回返回格式基本都遵循 OpenAI 的規(guī)范。每個 chunk 長這樣data: {choices:[{delta:{content:你},index:0}]} data: {choices:[{delta:{content:好},index:0}]} data: [DONE]注意delta字段它表示“增量內容”而不是完整內容。有些接口在第一個 chunk 里會返回role字段后續(xù) chunk 只有content。解析的時候要判斷delta.content是否存在不存在就跳過。我見過有人直接把delta整個渲染出來結果界面上出現(xiàn)了一堆{role:assistant}的 JSON 字符串非常尷尬。如果你用的是國內的大模型服務格式可能略有差異但核心思路一致每個 chunk 是一個 JSON里面有本次新增的文本片段。你需要把這些片段按順序拼接起來才能得到完整回答。5. 那些讓我加班到深夜的坑SSE 實戰(zhàn)排錯記錄5.1 消息被“攢”著一起發(fā)緩沖區(qū)的鍋這是最常見的問題沒有之一。本地開發(fā)時流式效果完美一部署到服務器就變成“一次性返回”。原因通常是中間有代理或網關開啟了緩沖。排查鏈路是這樣的先看響應頭有沒有X-Accel-Buffering: no沒有就加上。然后檢查 Nginx 配置proxy_buffering默認是on要改成off。如果用的是云服務商的負載均衡也要確認它有沒有對text/event-stream做特殊處理。最后檢查應用層有些框架的中間件會自動緩沖響應體比如某些日志中間件會等響應結束才記錄這就把流式給堵死了。提示排查流式問題時先用curl -N命令直接請求接口。-N參數會禁用 curl 的緩沖如果 curl 能看到逐塊輸出說明服務端沒問題問題出在瀏覽器到服務端之間的某一層。5.2 中文亂碼TextDecoder 的 stream 參數前面提過decoder.decode(value, { stream: true })里的stream: true這里再展開說一下。UTF-8 編碼的中文字符占 3 個字節(jié)網絡傳輸時可能把這三個字節(jié)切到兩個 chunk 里。如果每個 chunk 獨立解碼第一個 chunk 拿到不完整的字節(jié)序列就會解碼成亂碼。stream: true告訴解碼器“這不是最后一塊把不完整的字節(jié)緩存起來等下一塊來了再一起解”。這個參數不加中文場景必出問題。5.3 連接數限制HTTP/1.1 下的 6 連接瓶頸瀏覽器對同一域名的 HTTP/1.1 連接數限制通常是 6 個。SSE 連接是長連接會一直占著這個名額。如果你在頁面上同時開了多個 SSE 連接比如多個對話窗口第 7 個就會被阻塞。解決方案有幾個升級到 HTTP/2多路復用不受這個限制或者把 SSE 請求分散到不同子域名再或者用 WebSocket 替代。我在一個多標簽頁場景里遇到過這個問題用戶開了 7 個標簽頁第 7 個死活加載不出來排查了好久才定位到連接數限制。5.4 斷線重連EventSource 自動重連的副作用EventSource內置了自動重連連接斷了會按照retry指定的間隔重試。這個特性在普通場景下是優(yōu)點但在 AI 對話場景下可能是災難服務端正在生成回答網絡抖了一下EventSource自動重連服務端以為是新請求又從頭生成一遍用戶看到回答重復了。所以用EventSource時服務端要配合Last-Event-ID做斷點續(xù)傳或者前端在重連時主動帶上上下文標識讓服務端知道這是續(xù)傳而不是新請求。用fetch手動處理的話重連邏輯完全自己控制反而更省心。6. 把流式交互做得更順滑幾個提升體驗的細節(jié)6.1 前端渲染節(jié)奏別每個字符都觸發(fā)重排流式輸出時如果每收到一個字符就更新一次 DOM頁面會頻繁重排性能很差。我的做法是用一個緩沖區(qū)每隔 50 毫秒左右批量更新一次界面。這樣既保持了“打字機”的視覺效果又不會讓瀏覽器瘋狂重繪。具體實現(xiàn)可以用requestAnimationFrame或者簡單的定時器節(jié)流。另外Markdown 渲染在流式場景下要特別小心。如果每個 chunk 都重新解析整段 Markdown開銷很大而且未閉合的代碼塊會導致渲染錯亂。比較穩(wěn)妥的做法是流式過程中先用純文本展示等[DONE]之后再整體做一次 Markdown 渲染。或者用支持增量解析的 Markdown 庫但這類庫通常對未閉合語法的處理也不完美。6.2 錯誤處理流中斷了怎么給用戶交代流式請求比普通請求更容易中斷網絡波動、服務端超時、用戶主動取消都會導致流提前結束。前端需要區(qū)分幾種情況如果是用戶主動abort界面應該顯示“已停止生成”保留已經生成的內容如果是網絡錯誤應該顯示“連接中斷”并提供重試按鈕如果是服務端返回了錯誤信息要把錯誤內容展示出來。最忌諱的是流斷了但界面還在轉圈用戶完全不知道發(fā)生了什么。6.3 性能與成本流式不等于免費流式傳輸本身不增加太多服務器成本但它會讓連接保持更久。如果并發(fā)量大長連接會占用更多文件描述符和內存。另外前面提到客戶端中斷后服務端要及時停止生成否則大模型接口的調用費用照付。我在項目里加了一個“生成超時”機制超過 60 秒還沒生成完就強制結束避免異常情況下資源泄漏。7. 寫在最后一些個人體會流式傳輸和 SSE 這套東西剛接觸時覺得概念很多真正跑通一個 Demo 之后會發(fā)現(xiàn)核心就那么幾件事服務端按格式寫、客戶端按格式讀、中間別讓代理緩沖、中斷要前后端配合。難點不在協(xié)議本身而在各種環(huán)境下的兼容性和邊界情況處理。我自己的經驗是先把最簡單的EventSource版本跑通理解 SSE 的報文格式和事件機制然后再換成fetch手動解析把中斷、重連、錯誤處理這些補上。不要一上來就追求完美架構流式這東西調試成本不低先用最小可用版本驗證鏈路通暢再逐步加功能。還有一個建議多準備幾個調試工具。curl -N看服務端原始輸出瀏覽器開發(fā)者工具的 Network 面板看 SSE 流Chrome 對text/event-stream有專門的展示再配合服務端日志基本能覆蓋大部分問題。流式問題最怕的就是“黑盒”你不知道數據卡在哪一層有了這幾個工具排查效率會高很多。