解密 SSE，像 ChatGPT 一樣返回流式響應(yīng)
我們知道目前的 HTTP/1.1 采用的是標(biāo)準(zhǔn)的請(qǐng)求-響應(yīng)模型，客戶端主動(dòng)發(fā)請(qǐng)求，服務(wù)端被動(dòng)地返回響應(yīng)。這種模型在客戶端需要實(shí)時(shí)獲取結(jié)果的場(chǎng)景下是不合適的，因?yàn)檫@意味著客戶端需要不斷地輪詢，所以最好的做法是服務(wù)端生成結(jié)果之后，主動(dòng)推送給客戶端。
比如 ChatGPT，它在生成內(nèi)容時(shí)，也是生成一部分，就主動(dòng)向客戶端推送一部分。而在這個(gè)過(guò)程中，客戶端不需要做任何事情，只需等待 ChatGPT 服務(wù)端返回內(nèi)容即可。
說(shuō)到這兒，你肯定想到了 WebSocket，沒(méi)錯(cuò)這是一種解決方案。但 WebSocket 太重了，它和 HTTP 都是基于 TCP 的應(yīng)用層傳輸協(xié)議，只不過(guò)在握手的時(shí)候搭了 HTTP 的便車(chē)，利用 HTTP 本身的協(xié)議升級(jí)特性，偽裝成 HTTP，這樣就能繞過(guò)瀏覽器沙箱、網(wǎng)絡(luò)防火墻等限制。
當(dāng)完成握手之后，后續(xù)傳輸?shù)臄?shù)據(jù)就不再是 HTTP 報(bào)文，而是 WebSocket 格式的二進(jìn)制幀。所以這兩者完全是不同的協(xié)議，那有沒(méi)有一種辦法，我們?nèi)匀皇褂?HTTP 協(xié)議，同時(shí)還能讓服務(wù)端主動(dòng)推送數(shù)據(jù)呢？
答案是有的，也就是本文將要介紹的 SSE 技術(shù)，它的英文全稱(chēng)是 Server-Sent Events（服務(wù)端推送事件）。通過(guò) SSE 可以讓服務(wù)端即時(shí)推送數(shù)據(jù)到客戶端，而不需要客戶端輪詢服務(wù)端以獲取更新。
圖片
到這你可能會(huì)問(wèn)，那 WebSocket 和 SSE 有什么區(qū)別呢？
1）通信方式
WebSocket 提供全雙工通信，服務(wù)端和客戶端都可以在同一個(gè)連接上同時(shí)發(fā)送和接收數(shù)據(jù)。最重要的是，WebSocket 獨(dú)立于 HTTP 協(xié)議，盡管它開(kāi)始于一個(gè) HTTP 握手。
SSE 僅提供服務(wù)端到客戶端的單向通信，客戶端不能通過(guò) SSE 給服務(wù)端發(fā)信息。
2）協(xié)議和實(shí)現(xiàn)
WebSocket 使用自己的協(xié)議（ws:// 或 wss://），需要服務(wù)端和客戶端都支持，并且協(xié)議比較復(fù)雜。
SSE 則是使用標(biāo)準(zhǔn)的 HTTP 協(xié)議，實(shí)現(xiàn)起來(lái)更簡(jiǎn)單，尤其是在服務(wù)端。
3）適用場(chǎng)景
WebSocket 適用于服務(wù)端和客戶端之間雙向?qū)崟r(shí)通信的場(chǎng)景，如在線游戲、聊天應(yīng)用等。
SSE 適用于服務(wù)端向客戶端單向推送數(shù)據(jù)的場(chǎng)景，如消息通知、數(shù)據(jù)更新。并且 SSE 自動(dòng)支持?jǐn)嗑€重連，而 WebSocket 則需要額外部署。
4）復(fù)雜性和資源使用
WebSocket 由于其雙向通信的能力，通常比 SSE 更復(fù)雜，可能需要更多的資源來(lái)維護(hù)和管理連接。
SSE 因?yàn)槠鋯蜗蛐院突?HTTP 的特性，它可以利用現(xiàn)有的網(wǎng)絡(luò)基礎(chǔ)設(shè)施，如代理服務(wù)器、負(fù)載均衡器和防火墻等等，通常更容易實(shí)現(xiàn)和維護(hù)。
相信現(xiàn)在你已經(jīng)明白 SSE 是做什么的了，它的目的就是讓服務(wù)端能夠主動(dòng)推送數(shù)據(jù)給客戶端。如果不需要和服務(wù)端動(dòng)態(tài)交互，只是希望服務(wù)端在有數(shù)據(jù)的時(shí)候推過(guò)來(lái)，那么 WebSocket 就有些太重了，因?yàn)檫@意味著要替換 HTTP 協(xié)議，而使用 SSE 無(wú)疑是更好的選擇。
SSE 是什么我們已經(jīng)知道了，那它是怎么實(shí)現(xiàn)的呢？原理是什么呢？
1）建立連接
客戶端發(fā)起一個(gè)標(biāo)準(zhǔn)的 HTTP 請(qǐng)求來(lái)開(kāi)啟 SSE 會(huì)話，這個(gè)請(qǐng)求的特殊之處在于它包含一個(gè)頭字段。
Accept: text/event-stream相當(dāng)于客戶端告訴服務(wù)端，期望接收 SSE 消息流。而服務(wù)端在看到該字段時(shí)，也知道這是一個(gè) SSE 請(qǐng)求，于是立即向客戶端返回響應(yīng)頭，注意：返回的只有響應(yīng)頭，里面會(huì)包含如下頭字段。
Content-Type: text/event-stream響應(yīng)頭返回之后標(biāo)志著 SSE 連接成功建立，并且連接會(huì)保持開(kāi)放狀態(tài)，服務(wù)端后續(xù)可以隨時(shí)通過(guò)此連接向客戶端發(fā)送數(shù)據(jù)。此外當(dāng)連接不小心斷開(kāi)時(shí)，客戶端也會(huì)自動(dòng)進(jìn)行重連。
所以在普通的 HTTP 請(qǐng)求中，一旦服務(wù)端返回，那么請(qǐng)求結(jié)束了。雖然可以將 Connection 頭字段設(shè)置為 keep-alive 保證連接不斷開(kāi)，但每次訪問(wèn)都包含了 HTTP 請(qǐng)求/響應(yīng)的完整過(guò)程。
而在 SSE 中，服務(wù)端會(huì)保持一個(gè)開(kāi)放的連接，只要有新數(shù)據(jù)可用，就會(huì)直接發(fā)送給客戶端。所以服務(wù)端會(huì)將響應(yīng)以流的形式發(fā)送給客戶端，每次發(fā)送的消息都是響應(yīng)流的一部分，而不是獨(dú)立的 HTTP 響應(yīng)。
因此 SSE 的服務(wù)端在發(fā)送數(shù)據(jù)時(shí)，并不遵循傳統(tǒng)的一次請(qǐng)求，一次響應(yīng)模式。它在建立連接之后會(huì)保持連接開(kāi)放，并通過(guò)這個(gè)持續(xù)的連接流式地發(fā)送數(shù)據(jù)，這種方式就使得 SSE 非常適合實(shí)時(shí)數(shù)據(jù)推送的場(chǎng)景。
2）發(fā)送消息
客戶端發(fā)送請(qǐng)求，服務(wù)端返回響應(yīng)頭之后，SSE 連接就建立成功了。此時(shí)客戶端只需要躺平，安靜地等待服務(wù)端的輸出即可。所以現(xiàn)在的關(guān)鍵就在于服務(wù)端要返回什么格式的數(shù)據(jù)呢？很簡(jiǎn)單，一個(gè)基本的消息由以下幾部分組成：
- data：實(shí)際的消息數(shù)據(jù)；
- id：可選，消息的唯一標(biāo)識(shí)符，用于在連接重新建立時(shí)同步消息；
- event：可選，定義事件類(lèi)型，用于客戶端區(qū)分消息的類(lèi)型；
- retry：可選，自動(dòng)重連的時(shí)間（毫秒），如果連接中斷，客戶端在自動(dòng)重新連接之前，需要等待多長(zhǎng)時(shí)間；
注意：每個(gè)消息要以兩個(gè)換行符（\n\n）結(jié)束，舉個(gè)例子，我們發(fā)送一個(gè) Hello World。
data: Hello World\n\n也可以發(fā)送帶有事件類(lèi)型的消息：
event: userUpdate
data: {"username": "Serpen", "age": 18}\n\n還是比較簡(jiǎn)單的，服務(wù)端可以保持連接并隨時(shí)發(fā)送更多數(shù)據(jù)。然后客戶端在收到時(shí)會(huì)進(jìn)行處理，但不需要（也不能）對(duì)服務(wù)端作出任何回應(yīng)，它只需要被動(dòng)地接收來(lái)自服務(wù)端的數(shù)據(jù)即可。當(dāng)服務(wù)端認(rèn)為數(shù)據(jù)已經(jīng)全部發(fā)送完畢、無(wú)需再發(fā)時(shí)，那么便可以主動(dòng)斷開(kāi)連接。
關(guān)于 SSE 的原理我們就解釋清楚了，下面來(lái)實(shí)際編程實(shí)現(xiàn)它，這里我們先使用原生的 asyncio 實(shí)現(xiàn) SSE。
import asyncio
from asyncio import StreamReader, StreamWriter
class SSE:
def __init__(self, host="0.0.0.0", port=9999):
self.host = host
self.port = port
@staticmethod
def parse_request_headers(data: bytes) -> dict:
"""
此函數(shù)負(fù)責(zé)從原始字節(jié)流中解析出請(qǐng)求頭
"""
headers = data.split(b"\r\n\r\n")[0].split(b"\r\n")
header_dict = {}
for header in headers[1:]:
key, val = header.decode("utf-8").split(":", 1)
header_dict[key.lower()] = val.strip()
return header_dict
async def handler_requests(self,
reader: StreamReader,
writer: StreamWriter):
"""
負(fù)責(zé)處理來(lái)自客戶端的請(qǐng)求
每來(lái)一個(gè)客戶端連接，就會(huì)基于此函數(shù)創(chuàng)建一個(gè)協(xié)程
并且自動(dòng)傳遞兩個(gè)參數(shù)：reader 和 writer
reader.read 負(fù)責(zé)讀取數(shù)據(jù)，等價(jià)于 socket.recv
writer.write 負(fù)責(zé)發(fā)送數(shù)據(jù)，等價(jià)于 socket.send
"""
# 獲取客戶端的請(qǐng)求報(bào)文，這里對(duì)請(qǐng)求方法、請(qǐng)求地址不做限制
data = await reader.readuntil(b"\r\n\r\n")
# 解析出請(qǐng)求頭
request_headers = self.parse_request_headers(data)
# 簡(jiǎn)單檢測(cè)一下 accept 字段，如果不是建立 SSE，那么直接關(guān)閉連接
if request_headers.get("accept") != "text/event-stream":
writer.close()
return await writer.wait_closed()
# 如果是 SSE 連接，那么返回響應(yīng)頭
response_header = (
b"HTTP/1.1 200 OK\r\n"
b"Content-Type: text/event-stream\r\n"
b"Cache-Control: no-cache\r\n"
b"Connection: keep-alive\r\n"
b'Access-Control-Allow-Origin: *\r\n'
b"\r\n"
)
writer.write(response_header)
await writer.drain()
# 然后便可以不斷地向客戶端返回?cái)?shù)據(jù)了
for _ in range(5):
# 每隔 1 秒返回?cái)?shù)據(jù)
data = "data: 高老師總能分享出好東西\r\n\r\n".encode("utf-8")
writer.write(data)
await writer.drain()
await asyncio.sleep(1)
# 數(shù)據(jù)傳輸完畢
writer.close()
await writer.wait_closed()
async def __create_server(self):
# 創(chuàng)建服務(wù)，第一個(gè)參數(shù)是一個(gè)回調(diào)函數(shù)
# 當(dāng)連接過(guò)來(lái)的時(shí)候就會(huì)根據(jù)此函數(shù)創(chuàng)建一個(gè)協(xié)程
# 后面是綁定的 ip 和 端口
server = await asyncio.start_server(self.handler_requests,
self.host,
self.port)
# 然后開(kāi)啟無(wú)限循環(huán)
async with server:
await server.serve_forever()
def run_server(self):
loop = asyncio.get_event_loop()
loop.run_until_complete(self.__create_server())
if __name__ == '__main__':
sse = SSE()
sse.run_server()服務(wù)端代碼編寫(xiě)完畢，下面編寫(xiě)前端代碼。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
<style>
#data {
font-weight: bold;
color: cadetblue;
font-size: large;
}
</style>
</head>
<body>
<h1>SSE Test</h1>
<div id="data"></div>
<script>
document.addEventListener("DOMContentLoaded", function () {
// 和服務(wù)端建立 SSE 連接
var eventSource = new EventSource("http://localhost:9999");
eventSource.onmessage = function (e) {
// 將數(shù)據(jù)渲染在 <div id="data"></div> 的內(nèi)部
var data = e.data + "\n";
document.getElementById('data').innerText += data;
};
eventSource.onerror = function (e) {
console.error('Error occurred:', e);
eventSource.close();
};
});
</script>
</body>
</html>代碼編寫(xiě)完畢，我們用瀏覽器打開(kāi) HTML 文件，便可看到如下效果。
圖片
以上我們就簡(jiǎn)單實(shí)現(xiàn)了 SSE，當(dāng)然為了加深印象，這里的后端是使用原生的 asyncio 編寫(xiě)的，但在工作中，我們會(huì)使用現(xiàn)成的 Web 框架，比如 FastAPI，Blacksheep 等等。
需要說(shuō)明的是，雖然通過(guò) SSE 技術(shù)可以實(shí)現(xiàn)類(lèi)似 ChatGPT 的效果，但 ChatGPT 內(nèi)部并沒(méi)有用到 SSE，它內(nèi)部是基于 HTTP 的分塊傳輸實(shí)現(xiàn)的。因?yàn)?SSE 只能通過(guò) GET 請(qǐng)求發(fā)出，并且無(wú)法自定義請(qǐng)求頭。
如果想實(shí)現(xiàn) ChatGPT 的效果，需要使用 HTTP 的分塊傳輸。而像 FastAPI、BlackSheep 等框架提供的流式響應(yīng)，便是基于 HTTP 的分塊傳輸實(shí)現(xiàn)的，比如 FastAPI：
import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
async def event_generator():
for _ in range(5):
# 每隔 1 秒返回?cái)?shù)據(jù)
data = "data: 高老師總能分享出好東西\r\n\r\n".encode("utf-8")
yield data
await asyncio.sleep(1)
@app.get("/")
async def sse():
return StreamingResponse(event_generator(),
media_type="text/event-stream")
if __name__ == '__main__':
uvicorn.run(app, host="0.0.0.0", port=9999)首先之前的前端代碼依舊可以正常訪問(wèn)，通過(guò)修改數(shù)據(jù)格式和 Content-Type 可以讓其支持 SSE。但最正確的做法是直接訪問(wèn) localhost:9999，效果如下：
圖片
所以基于 StreamingResponse 可以實(shí)現(xiàn) SSE，也可以直接訪問(wèn)。而直接訪問(wèn)的話，此時(shí)里面的 data: 和 \r\n 就是實(shí)體數(shù)據(jù)的一部分。并且這種方式和 ChatGPT 的工作機(jī)制是相似的，都使用了 HTTP 的分塊傳輸，支持所有的請(qǐng)求方法，而 SSE 只支持 GET 請(qǐng)求。
BlackSheep 也是類(lèi)似的，它同樣也支持流式響應(yīng)。
import asyncio
from blacksheep import Application, Response, StreamedContent
import uvicorn
app = Application()
app.use_cors(
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
async def event_generator():
for _ in range(5):
# 每隔 1 秒返回?cái)?shù)據(jù)
data = "data: 高老師總能分享出好東西\r\n\r\n".encode("utf-8")
yield data
await asyncio.sleep(1)
@app.router.get("/")
async def sse():
return Response(
200,
cnotallow=StreamedContent(b"text/event-stream", event_generator),
)
if __name__ == '__main__':
uvicorn.run(app, host="0.0.0.0", port=9999)可以測(cè)試一下，效果是一樣的。如果你不想實(shí)現(xiàn) SSE，只是希望固定的數(shù)據(jù)以流的形式一點(diǎn)一點(diǎn)返回，那么記得將數(shù)據(jù)中多余的 data: 和 \r\n 給去掉，并最好修改 Content-Type 為合適的類(lèi)型。
所以 SSE 一般用于需要服務(wù)端推數(shù)據(jù)，但數(shù)據(jù)不知道什么時(shí)候會(huì)過(guò)來(lái)，于是通過(guò) SSE 保持連接開(kāi)放。后續(xù)當(dāng)服務(wù)端有數(shù)據(jù)了，直接通過(guò)連接發(fā)送給客戶端即可。
而 FastAPI 和 BlackSheep 提供的流式響應(yīng)更像是，返回的數(shù)據(jù)比較龐大，如果全部準(zhǔn)備好再一次性返回，會(huì)讓用戶陷入長(zhǎng)時(shí)間的等待，造成不好的體驗(yàn)。于是通過(guò)分塊傳輸，準(zhǔn)備好一部分就返回一部分。雖然整體時(shí)間沒(méi)變，但可以讓用戶立刻獲取到數(shù)據(jù)，從而提升用戶體驗(yàn)。
比如 ChatGPT，當(dāng)它回答的內(nèi)容比較多的時(shí)候，那么整個(gè)過(guò)程耗費(fèi)幾十秒鐘是常有的事情，假設(shè) 30 秒。相比讓用戶等待 30 秒，然后內(nèi)容一下子刷出來(lái)，顯然生成一部分返回一部分這種方式更讓人喜歡。
因此使用 SSE 還是流式響應(yīng)，則取決于你當(dāng)前的業(yè)務(wù)。如果你返回的數(shù)據(jù)是確定的，只是準(zhǔn)備的時(shí)間比較長(zhǎng)，或者數(shù)據(jù)量比較大，那么推薦使用流式響應(yīng)。
至于 SSE，在這些現(xiàn)成的 Web 框架里面，也可以通過(guò)流式響應(yīng)來(lái)實(shí)現(xiàn)，只需要將 Content-Type 設(shè)置為 text/event-stream，并將數(shù)據(jù)加上前綴 data: 和后綴 \r\n\r\n。
但說(shuō)實(shí)話，如果想實(shí)現(xiàn) SSE，不建議通過(guò)流式響應(yīng)來(lái)實(shí)現(xiàn)，而是使用專(zhuān)門(mén)的庫(kù)。以 FastAPI 為例：
from sse_starlette.sse import EventSourceResponseFastAPI 其實(shí)就是在 starlette 的基礎(chǔ)上套了一層殼，通過(guò)安裝 sse_starlette 可以讓 FastAPI 更好地支持 SSE。
以上就是本文的內(nèi)容，如果對(duì)你有幫助，就點(diǎn)個(gè)贊吧。
