1、前言

如果項(xiàng)目中有一個(gè)場(chǎng)景，假設(shè)對(duì)接ChatGPT或?qū)犹鞖忸惤涌诘臅r(shí)候，需要服務(wù)端主動(dòng)往客戶端進(jìn)行消息推送或推流。通常的做法有：

• 客戶端提供接收數(shù)據(jù)接口，服務(wù)端開(kāi)啟定時(shí)輪詢，定時(shí)向客戶端發(fā)起http請(qǐng)求
• 客戶端提供定時(shí)輪詢服務(wù)，定時(shí)向服務(wù)端發(fā)起http請(qǐng)求接口
• 使用websocket實(shí)時(shí)通訊

那么今天再介紹另一種機(jī)制：SSE，也就是服務(wù)器發(fā)送事件機(jī)制。

2、什么是SSE

SSE（Server-Sent Events）是一種允許服務(wù)器向客戶端推送實(shí)時(shí)數(shù)據(jù)的技術(shù)，它建立在 HTTP 和簡(jiǎn)單文本格式之上，提供了一種輕量級(jí)的服務(wù)器推送方式，通常也被稱為“事件流”（Event Stream）。他通過(guò)在客戶端和服務(wù)端之間建立一個(gè)長(zhǎng)連接，并通過(guò)這條連接實(shí)現(xiàn)服務(wù)端和客戶端的消息實(shí)時(shí)推送。

2.1、技術(shù)原理

SSE是建立在HTTP協(xié)議之上的，所以原理比較簡(jiǎn)單，也與HTTP原理類似：

1）建立連接：

客戶端通過(guò)普通的 HTTP 請(qǐng)求向服務(wù)器發(fā)起連接請(qǐng)求，類似于普通的 Web 請(qǐng)求。這個(gè)請(qǐng)求的關(guān)鍵在于使用了 text/event-stream 的 MIME 類型，告知服務(wù)器該請(qǐng)求是 SSE 請(qǐng)求。

httpCopy codeGET /sse/stream HTTP/1.1
Host: example.com
Accept: text/event-stream

2）服務(wù)器處理請(qǐng)求：

服務(wù)器接收到 SSE 請(qǐng)求后，會(huì)在連接上保持打開(kāi)狀態(tài)，不會(huì)立即關(guān)閉。這是與普通的請(qǐng)求-響應(yīng)模式的主要不同之處。服務(wù)器端通過(guò)這個(gè)持久連接向客戶端發(fā)送數(shù)據(jù)。

3）數(shù)據(jù)推送：

服務(wù)器端通過(guò)打開(kāi)的連接，周期性地向客戶端發(fā)送消息。這些消息以文本的形式發(fā)送，并遵循一定的格式，通常以 data 字段表示消息內(nèi)容。

httpCopy codeHTTP/1.1 200 OK
Content-Type: text/event-stream
data: This is a message\n\n

上述例子中，data 字段包含了實(shí)際的消息內(nèi)容，兩個(gè)換行符（\n\n）表示消息的結(jié)束。

4）客戶端接收消息：

客戶端通過(guò)監(jiān)聽(tīng)連接的 message 事件來(lái)接收服務(wù)器推送的消息。一旦接收到消息，客戶端可以采取相應(yīng)的操作，例如更新界面內(nèi)容。

javascriptCopy codeconst eventSource = new EventSource('/sse/stream');
eventSource.onmessage = function (event) {
    console.log('Received message:', event.data);
    // 處理消息，例如更新界面
};

5）連接關(guān)閉：

當(dāng)服務(wù)器端不再需要向客戶端推送消息時(shí)，或者發(fā)生錯(cuò)誤時(shí)，服務(wù)器可以關(guān)閉連接?？蛻舳艘部梢酝ㄟ^(guò)調(diào)用 eventSource.close() 來(lái)關(guān)閉連接。

2.2、SSE和WebSocket

提到SSE，那自然要提一下WebSocket了。WebSocket是一種HTML5提供的全雙工通信協(xié)議（指可以在同一時(shí)間內(nèi)允許兩個(gè)設(shè)備之間進(jìn)行雙向發(fā)送和接收數(shù)據(jù)的通信協(xié)議），基于TCP協(xié)議，并復(fù)用HTTP的握手通道（允許一次TCP連接中傳輸多個(gè)HTTP請(qǐng)求和相應(yīng)），常用于瀏覽器與服務(wù)器之間的實(shí)時(shí)通信。

SSE和WebSocket盡管功能類似，都是用來(lái)實(shí)現(xiàn)服務(wù)器向客戶端實(shí)時(shí)推送數(shù)據(jù)的技術(shù)，但還是有一定區(qū)別：

2.2.1、SSE (Server-Sent Events)

• 簡(jiǎn)單性：SSE 使用簡(jiǎn)單的 HTTP 協(xié)議，通常建立在標(biāo)準(zhǔn)的 HTTP 或 HTTPS 連接之上。這使得它對(duì)于一些簡(jiǎn)單的實(shí)時(shí)通知場(chǎng)景非常適用，特別是對(duì)于服務(wù)器向客戶端單向推送數(shù)據(jù)。
• 兼容性：SSE 在瀏覽器端具有較好的兼容性，因?yàn)樗腔跇?biāo)準(zhǔn)的 HTTP 協(xié)議的。即使在一些不支持 WebSocket 的環(huán)境中，SSE 仍然可以被支持。
• 適用范圍：SSE 適用于服務(wù)器向客戶端單向推送通知，例如實(shí)時(shí)更新、事件通知等。但它僅支持從服務(wù)器到客戶端的單向通信，客戶端無(wú)法直接向服務(wù)器發(fā)送消息。

2.2.2、WebSocket

全雙工通信： WebSocket 提供了全雙工通信，允許客戶端和服務(wù)器之間進(jìn)行雙向?qū)崟r(shí)通信。這使得它適用于一些需要雙向數(shù)據(jù)交換的應(yīng)用，比如在線聊天、實(shí)時(shí)協(xié)作等。
低延遲：WebSocket 的通信開(kāi)銷相對(duì)較小，因?yàn)樗褂脝我坏某志眠B接，而不像 SSE 需要不斷地創(chuàng)建新的連接。這可以降低通信的延遲。
適用范圍： WebSocket 適用于需要實(shí)時(shí)雙向通信的應(yīng)用，特別是對(duì)于那些需要低延遲、高頻率消息交換的場(chǎng)景。

2.2.3、選擇 SSE 還是 WebSocket？

簡(jiǎn)單通知場(chǎng)景：如果你只需要服務(wù)器向客戶端推送簡(jiǎn)單的通知、事件更新等，而不需要客戶端與服務(wù)器進(jìn)行雙向通信，那么 SSE 是一個(gè)簡(jiǎn)單而有效的選擇。
雙向通信場(chǎng)景：如果你的應(yīng)用需要實(shí)現(xiàn)實(shí)時(shí)雙向通信，例如在線聊天、協(xié)作編輯等，那么 WebSocket 是更合適的選擇。
兼容性考慮： 如果你的應(yīng)用可能在一些不支持 WebSocket 的環(huán)境中運(yùn)行，或者需要考慮到更廣泛的瀏覽器兼容性，那么 SSE 可能是一個(gè)更可行的選擇。

3、Springboot快速集成

3.1、添加依賴

Springboot項(xiàng)目中，sse不需要額外添加依賴，引用了web相關(guān)的springboot依賴即可：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

3.2、創(chuàng)建SSE控制器

這里簡(jiǎn)單創(chuàng)建一個(gè)控制器類，用于處理SSE請(qǐng)求。在JAVA中通常使用SSEmitter來(lái)實(shí)現(xiàn)sse的消息推送。

package com.example.springbootsse.controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Date;
@RestController
@RequestMapping("/sse")
public class SSEmitterController {
    @GetMapping("/stream")
    public SseEmitter stream() {
        // 用于創(chuàng)建一個(gè) SSE 連接對(duì)象
        SseEmitter emitter = new SseEmitter();
        // 在后臺(tái)線程中模擬實(shí)時(shí)數(shù)據(jù)
        new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    // emitter.send() 方法向客戶端發(fā)送消息
                    // 使用SseEmitter.event()創(chuàng)建一個(gè)事件對(duì)象，設(shè)置事件名稱和數(shù)據(jù)
                    emitter.send(SseEmitter.event().name("message").data("[" + new Date() + "] Data #" + i));
                    Thread.sleep(1000);
                }
                // 數(shù)據(jù)發(fā)送完成后，關(guān)閉連接
                emitter.complete(); 
            } catch (IOException | InterruptedException e) {
                 // 發(fā)生錯(cuò)誤時(shí)，關(guān)閉連接并報(bào)錯(cuò)
                emitter.completeWithError(e);
            }
        }).start();
        return emitter;
    }
}

查看執(zhí)行結(jié)果，可以看到每一秒服務(wù)端都會(huì)自動(dòng)像客戶端推送messag消息：

我們來(lái)關(guān)注下SSEmitter這個(gè)類，SseEmitter 是 Spring Framework 中用于實(shí)現(xiàn) Server-Sent Events（SSE）的一個(gè)類。它允許服務(wù)器向客戶端推送數(shù)據(jù)，通過(guò)建立一個(gè)持久連接，實(shí)現(xiàn)服務(wù)器向客戶端的實(shí)時(shí)單向通信。在 Spring 框架中，SseEmitter 類通常用于處理 SSE 請(qǐng)求，推送事件給客戶端。

3.2.1、SSEmitter創(chuàng)建實(shí)例

SSEmitter提供了兩個(gè)構(gòu)造函數(shù)用于創(chuàng)建實(shí)例。在創(chuàng)建實(shí)例時(shí)，我們可以指定超時(shí)時(shí)間timeout，如果傳0或使用無(wú)參構(gòu)造，則表示永不過(guò)期。連接超時(shí)是指在一段時(shí)間內(nèi)沒(méi)有數(shù)據(jù)傳輸時(shí)，連接將被認(rèn)為是超時(shí)的，并自動(dòng)關(guān)閉。

3.2.2、SSEmitter API

除此以外，SSEmitter還提供了幾種API，如上面例子中使用到的：

• emitter.send() 方法向客戶端發(fā)送消息。
• SseEmitter.event() 創(chuàng)建一個(gè)事件對(duì)象，設(shè)置事件名稱和數(shù)據(jù)。
• emitter.complete() 表示數(shù)據(jù)發(fā)送完成后關(guān)閉連接。
• emitter.completeWithError(e) 在發(fā)生錯(cuò)誤時(shí)關(guān)閉連接并報(bào)錯(cuò)。

3.2.3、SSEmitter注冊(cè)回調(diào)

SseEmitter 可以通過(guò)注冊(cè)回調(diào)函數(shù)來(lái)處理服務(wù)器端發(fā)往客戶端的事件。當(dāng)服務(wù)器端有新的數(shù)據(jù)需要推送給客戶端時(shí)，注冊(cè)的回調(diào)函數(shù)將會(huì)被調(diào)用。SSEmitter繼承了ResponseBodyEmitter，提供的一系列注冊(cè)回調(diào)函數(shù)有：

示例代碼：

package com.example.springbootsse.controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;
import java.util.Date;
@RestController
@RequestMapping("/sse")
public class SSEmitterController {
    @GetMapping("/stream")
    public SseEmitter stream() {
        // 3S超時(shí)
        SseEmitter emitter = new SseEmitter(10000L);
        // 注冊(cè)回調(diào)函數(shù)，處理服務(wù)器向客戶端推送的消息
        emitter.onCompletion(() -> {
            System.out.println("Connection completed");
            // 在連接完成時(shí)執(zhí)行一些清理工作
        });
        emitter.onTimeout(() -> {
            System.out.println("Connection timeout");
            // 在連接超時(shí)時(shí)執(zhí)行一些處理
            emitter.complete();
        });
        // 在后臺(tái)線程中模擬實(shí)時(shí)數(shù)據(jù)
        new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    emitter.send(SseEmitter.event().name("message").data("[" + new Date() + "] Data #" + i));
                    Thread.sleep(1000);
                }
                emitter.complete(); // 數(shù)據(jù)發(fā)送完成后，關(guān)閉連接
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e); // 發(fā)生錯(cuò)誤時(shí)，關(guān)閉連接并報(bào)錯(cuò)
            }
        }).start();
        return emitter;
    }
}

• onCompletion()：在連接完成時(shí)候觸發(fā)，可在連接完成時(shí)執(zhí)行一些清理工作
• onTimeout()：當(dāng)連接超時(shí)時(shí)觸發(fā)
• onError()：當(dāng)連接異常時(shí)觸發(fā)
• completeWithError(e)：用于發(fā)生錯(cuò)誤時(shí)，關(guān)閉連接并報(bào)錯(cuò)

4、小結(jié)

其實(shí)SSE已經(jīng)出來(lái)很久了，但是熟知他的人卻很少，大多數(shù)項(xiàng)目中還是直接使用了websocket技術(shù)。直到最近ChatGPT火了之后，很多項(xiàng)目需要對(duì)接GPT進(jìn)行實(shí)時(shí)推流，才逐漸又被人提起。所以借此篇文章給自己掃盲一下。

到此這篇關(guān)于Springboot集成sse服務(wù)端推流的文章就介紹到這了,更多相關(guān)Springboot集成sse內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論