在現(xiàn)代 Web 應(yīng)用中，實(shí)時(shí)數(shù)據(jù)傳輸已經(jīng)成為一個(gè)不可忽視的需求，尤其是在聊天系統(tǒng)、實(shí)時(shí)數(shù)據(jù)展示、進(jìn)度推送等場(chǎng)景中。

Spring 提供了 SseEmitter 來(lái)支持服務(wù)器端推送事件（Server-Sent Events, SSE），它通過(guò) HTTP 協(xié)議允許服務(wù)器主動(dòng)向客戶端推送數(shù)據(jù)。與 WebSocket 不同，SSE 是單向通信，但它也提供了一種輕量、高效的實(shí)時(shí)數(shù)據(jù)流解決方案。

本文將從原理、實(shí)際應(yīng)用、客戶端和服務(wù)端代碼示例、擴(kuò)展應(yīng)用場(chǎng)景等方面詳細(xì)介紹 Spring SseEmitter。

1. 原理解析

1.1 Server-Sent Events (SSE)

SSE 是一種基于 HTTP 協(xié)議的單向數(shù)據(jù)流技術(shù)，允許服務(wù)器通過(guò)持久化的 HTTP 連接將事件推送到客戶端。其與 WebSocket 的主要區(qū)別是，SSE 是單向通信（從服務(wù)器到客戶端），而 WebSocket 是全雙工通信（服務(wù)器和客戶端都可以發(fā)送數(shù)據(jù)）。

在 SSE 中，客戶端通過(guò) EventSource 對(duì)象與服務(wù)器建立連接，服務(wù)器將數(shù)據(jù)以流的方式推送到客戶端。SSE 使用標(biāo)準(zhǔn)的 HTTP 協(xié)議，并且是基于文本的流，數(shù)據(jù)通常以 text/event-stream 類型的格式傳輸。

1.2 SseEmitter

SseEmitter 是 Spring 提供的一個(gè)類，用于實(shí)現(xiàn) SSE 功能。它支持向客戶端推送數(shù)據(jù)，并且能夠處理長(zhǎng)連接。SseEmitter 的主要功能包括：

• 異步發(fā)送數(shù)據(jù)：SSE 是長(zhǎng)連接，SseEmitter 可以在服務(wù)器端異步發(fā)送數(shù)據(jù)。
• 支持超時(shí)管理：SseEmitter 可以設(shè)置超時(shí)，若客戶端在指定時(shí)間內(nèi)沒有響應(yīng)，連接會(huì)自動(dòng)關(guān)閉。
• 異常處理：如果發(fā)生錯(cuò)誤，可以通過(guò) SseEmitter 完成錯(cuò)誤通知和連接關(guān)閉。

2. 服務(wù)端代碼實(shí)現(xiàn)

2.1 創(chuàng)建 SSE 連接

使用 Spring 的 SseEmitter 來(lái)向客戶端推送實(shí)時(shí)事件非常簡(jiǎn)單。

我們通過(guò) @GetMapping 注解來(lái)映射一個(gè)請(qǐng)求路徑，返回一個(gè) SseEmitter 實(shí)例。

@RestController
public class SseController {

    @GetMapping(value = "/sync/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter syncStream() {
        SseEmitter emitter = new SseEmitter();

        // 模擬異步任務(wù)，發(fā)送實(shí)時(shí)數(shù)據(jù)
        new Thread(() -> {
            try {
                for (int i = 0; i <= 100; i++) {
                    emitter.send("data: " + i + "% complete\n\n");
                    Thread.sleep(1000);  // 模擬任務(wù)處理
                }
                emitter.complete();  // 任務(wù)完成，結(jié)束 SSE 流
            } catch (Exception e) {
                emitter.completeWithError(e);  // 出現(xiàn)異常時(shí)結(jié)束流
            }
        }).start();

        return emitter;
    }
}

在上面的代碼中，syncStream 方法創(chuàng)建了一個(gè) SseEmitter 實(shí)例，并使用一個(gè)線程模擬了一個(gè)處理任務(wù)的過(guò)程。

每秒鐘將當(dāng)前進(jìn)度推送給客戶端，直到進(jìn)度達(dá)到 100%。當(dāng)任務(wù)完成時(shí)，通過(guò) emitter.complete() 關(guān)閉連接。

• SseEmitter.send(data): 用于向客戶端推送數(shù)據(jù)。
• SseEmitter.complete(): 用于標(biāo)記推送完成，關(guān)閉連接。
• SseEmitter.completeWithError(exception): 如果發(fā)生異常，可以通過(guò)此方法結(jié)束連接并發(fā)送錯(cuò)誤信息。

2.2 超時(shí)設(shè)置

SSE 連接可能會(huì)因?yàn)榫W(wǎng)絡(luò)問(wèn)題或其他原因被中斷，Spring 提供了超時(shí)設(shè)置，確保連接不被無(wú)限期阻塞。

@GetMapping(value = "/sync/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter syncStream() {
    SseEmitter emitter = new SseEmitter(30000L);  // 設(shè)置超時(shí)時(shí)間為30秒
    // 發(fā)送數(shù)據(jù)的代碼同上...
    return emitter;
}

3. 客戶端代碼實(shí)現(xiàn)

在客戶端，我們可以使用原生的 JavaScript EventSource 對(duì)象來(lái)接收 SSE 流數(shù)據(jù)。EventSource 會(huì)自動(dòng)管理連接，包括重試和重新連接等操作。

let eventSource = new EventSource("/sync/stream");

eventSource.onmessage = function(event) {
    console.log("Received message: ", event.data);
};

eventSource.onerror = function(error) {
    console.error("EventSource failed: ", error);
};

• new EventSource("/sync/stream"): 創(chuàng)建一個(gè)連接到指定 URL 的 EventSource 實(shí)例。
• onmessage: 每當(dāng)服務(wù)器發(fā)送一個(gè)事件時(shí)，會(huì)觸發(fā)該事件。
• onerror: 當(dāng)發(fā)生錯(cuò)誤時(shí)，會(huì)觸發(fā)該事件。

4. 擴(kuò)展應(yīng)用場(chǎng)景

4.1 實(shí)時(shí)任務(wù)進(jìn)度推送

可以利用 SseEmitter 來(lái)推送后臺(tái)任務(wù)的實(shí)時(shí)進(jìn)度。

例如，在執(zhí)行一個(gè)需要時(shí)間的任務(wù)時(shí)，可以定期向客戶端推送進(jìn)度更新。

@GetMapping("/task/progress")
public SseEmitter streamTaskProgress() {
    SseEmitter emitter = new SseEmitter();
    new Thread(() -> {
        try {
            for (int i = 0; i <= 100; i++) {
                emitter.send("data: " + i + "% complete\n\n");
                Thread.sleep(1000);  // 模擬任務(wù)處理
            }
            emitter.complete();  // 任務(wù)完成，結(jié)束 SSE 流
        } catch (Exception e) {
            emitter.completeWithError(e);  // 出現(xiàn)異常時(shí)結(jié)束流
        }
    }).start();
    return emitter;
}

4.2 實(shí)時(shí)聊天系統(tǒng)

在實(shí)時(shí)聊天應(yīng)用中，可以使用 SSE 向所有連接的客戶端推送消息。

@GetMapping("/chat/{roomId}")
public SseEmitter streamChatMessages(@PathVariable String roomId) {
    SseEmitter emitter = new SseEmitter();
    chatService.addListener(roomId, message -> {
        try {
            emitter.send("data: " + message + "\n\n");
        } catch (IOException e) {
            emitter.completeWithError(e);
        }
    });
    return emitter;
}

在這個(gè)例子中，

chatService.addListener(roomId, message -> {...}) 

是監(jiān)聽指定聊天室消息的邏輯，所有聊天信息都會(huì)通過(guò) SseEmitter.send() 推送到客戶端。

4.3 動(dòng)態(tài)數(shù)據(jù)顯示

SSE 適用于動(dòng)態(tài)顯示數(shù)據(jù)更新的場(chǎng)景，比如股票價(jià)格、天氣預(yù)報(bào)等實(shí)時(shí)數(shù)據(jù)。

@GetMapping("/live-stock/{symbol}")
public SseEmitter streamStockPrice(@PathVariable String symbol) {
    SseEmitter emitter = new SseEmitter();
    stockService.addPriceListener(symbol, price -> {
        try {
            emitter.send("data: " + price + "\n\n");
        } catch (IOException e) {
            emitter.completeWithError(e);
        }
    });
    return emitter;
}

在這個(gè)例子中，

stockService.addPriceListener(symbol, price -> {...})

監(jiān)聽股票的實(shí)時(shí)價(jià)格變化，并通過(guò) SseEmitter.send() 將最新價(jià)格推送到客戶端。

5. 總結(jié)

Spring 的 SseEmitter 提供了一種簡(jiǎn)潔且高效的方式來(lái)實(shí)現(xiàn)服務(wù)器向客戶端推送實(shí)時(shí)事件。通過(guò) SSE，我們可以輕松地在后臺(tái)執(zhí)行長(zhǎng)時(shí)間任務(wù)，并將實(shí)時(shí)數(shù)據(jù)推送給前端應(yīng)用。

SSE 適用于許多場(chǎng)景，如任務(wù)進(jìn)度推送、實(shí)時(shí)消息通知、動(dòng)態(tài)數(shù)據(jù)展示等。相較于 WebSocket，SSE 實(shí)現(xiàn)簡(jiǎn)單，且不需要額外的協(xié)議支持，適合輕量級(jí)的實(shí)時(shí)應(yīng)用。

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評(píng)論