腳本之家服務器常用軟件

快捷導航

SpringBoot3中Spring?WebFlux?SSE服務器發(fā)送事件的實現(xiàn)步驟

更新時間：2024年11月21日 14:12:28 作者：CoderJia

本文介紹了如何使用SpringBoot3和響應式編程實現(xiàn)服務器發(fā)送事件（SSE）,并討論了其在實時數(shù)據推送場景中的優(yōu)勢,通過示例代碼,展示了如何創(chuàng)建SSE控制器、客戶端接收數(shù)據以及優(yōu)化與擴展,感興趣的朋友跟隨小編一起看看吧

ChatGPT 剛出的時候，讓大伙很好奇的是它是如何實現(xiàn)的逐字輸出的？答案就是 SSE （服務器發(fā)送事件）。隨著實時數(shù)據和響應式編程的需求不斷增加，服務器發(fā)送事件（Server-Sent Events，簡稱 SSE）在現(xiàn)代 Web 應用程序中越來越受歡迎。SSE 提供了一種輕量級的服務器推送數(shù)據給客戶端的方式，適合用于監(jiān)控、實時通知、股票價格更新等場景。

在 Spring Boot 3 中，結合響應式編程的理念，SSE 的實現(xiàn)變得更加簡潔和高效。本文將詳細介紹如何使用 Spring Boot 3 來實現(xiàn) SSE 服務端推送，并討論響應式編程在此過程中的重要性和優(yōu)勢。

1. 什么是 SSE？

服務器發(fā)送事件（SSE） 是一種從服務器向客戶端推送數(shù)據的技術，屬于 HTML5 的一部分。與傳統(tǒng)的 HTTP 請求-響應模型不同，SSE 是單向的，服務器可以持續(xù)不斷地向客戶端發(fā)送數(shù)據，而客戶端通過一次長連接持續(xù)接收這些更新。

相比 WebSocket，SSE 有以下特點：

單向通信：SSE 僅允許服務器向客戶端推送數(shù)據，客戶端無法向服務器發(fā)送數(shù)據。
基于 HTTP 協(xié)議：SSE 是建立在 HTTP 協(xié)議之上的，瀏覽器原生支持，不需要額外的協(xié)議處理。
自動重連：SSE 支持自動重連，當連接意外斷開時，客戶端會自動嘗試重新連接服務器。

2. Spring Boot 3 響應式編程與 SSE

Spring Boot 3 提供了對響應式編程的全面支持，基于 Project Reactor 實現(xiàn)異步、非阻塞的流式數(shù)據處理。而響應式編程非常適合實現(xiàn) SSE，因為它允許我們以非阻塞的方式持續(xù)推送數(shù)據，而不會阻塞服務器的資源。

Spring WebFlux 是 Spring Boot 3 中用于構建響應式應用的核心框架，它可以無縫集成 SSE，為我們提供簡單高效的服務器推送功能。

為什么選擇響應式編程實現(xiàn) SSE？

傳統(tǒng)的阻塞式編程在處理長連接（如 SSE）時可能會占用大量服務器資源。響應式編程通過非阻塞 I/O 操作，不僅可以高效處理長時間的連接，還能在有新數(shù)據時立即推送給客戶端。響應式流（如 Flux）天然適合于這種流式數(shù)據推送場景。

3. 實現(xiàn) SSE 的基本步驟

我們將通過以下步驟實現(xiàn)一個簡單的 SSE 服務端推送應用：

創(chuàng)建 Spring Boot 項目并引入 WebFlux 依賴。
實現(xiàn)服務端推送 SSE 事件流。
編寫客戶端接收 SSE 數(shù)據。
測試與優(yōu)化。

3.1 創(chuàng)建 Spring Boot 項目

首先，創(chuàng)建一個新的 Spring Boot 3 項目，并確保引入了 spring-boot-starter-webflux 依賴。可以通過 Maven 或 Gradle 配置：

Maven 依賴

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

3.2 實現(xiàn)服務端推送SSE 事件流

在 Spring WebFlux 中，SSE 通過返回 Flux<ServerSentEvent<T>> 這種響應流來實現(xiàn)。下面我們實現(xiàn)一個簡單的 SSE 控制器，它會每隔一段時間向客戶端推送當前的時間信息。

示例控制器

package com.coderjia.boot3webflux.controller;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.time.LocalTime;
/**
 * @author CoderJia
 * @create 2024/10/27 下午 07:03
 * @Description
 **/
@Controller
public class SseController {
    @GetMapping("/sse/stream")
    public Flux<ServerSentEvent<String>> streamSse() {
        return Flux.interval(Duration.ofSeconds(1))
                .map(sequence -> ServerSentEvent.<String>builder()
                        .id(String.valueOf(sequence))
                        .event("periodic-event")
                        .data("Current time: " + LocalTime.now())
                        .build());
    }
}

解釋

Flux.interval(Duration.ofSeconds(1))：創(chuàng)建一個每秒發(fā)出事件的響應式流。

ServerSentEvent.builder()：構建 ServerSentEvent 對象，它可以包含 id、event 和 data 等信息，符合 SSE 規(guī)范。

map()：將流中的每個事件映射為 ServerSentEvent，并附帶當前的時間信息。

3.3 客戶端接收 SSE 數(shù)據

客戶端可以使用 JavaScript 原生的 EventSource API 來接收服務器發(fā)送的 SSE 數(shù)據流。

示例 HTML + JavaScript 客戶端

resources/static/index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SSE Example</title>
</head>
<body>
<h1>Server-Sent Events (SSE) Example</h1>
<div id="messages"></div>
<script src="https://unpkg.com/axios/dist/axios.min.js"></script>
<script>
    const http = axios.create({
        baseURL: 'http://localhost:8080/',
        timeout: 100000,
        responseType: 'stream',
        onDownloadProgress: function (progressEvent) {
            // 獲取 messages 元素
            const messagesElement = document.getElementById("messages");
            // 清除現(xiàn)有內容
            messagesElement.innerHTML = "";
            // 添加新內容
            const newElement = document.createElement("div");
            newElement.innerHTML = progressEvent.event.currentTarget.responseText + "<br/>";
            messagesElement.appendChild(newElement);
        },
    });
    http.get('/sse/stream')
        .then(function (response) {
            // 處理成功情況
            console.log(response);
        })
        .catch(function (error) {
            // 處理錯誤情況
            console.log(error);
        })
        .finally(function () {
            // 總是會執(zhí)行
        });
</script>
</body>
</html>

解釋

EventSource("/sse/stream")：EventSource 是瀏覽器提供的一個用于和服務器建立連接，接收服務器發(fā)送事件的接口。在客戶端發(fā)起與服務器的 SSE 長連接。服務器通過 /sse/stream 推送事件。
onmessage：處理服務器發(fā)送的消息，并將消息顯示在頁面上。
onerror：當連接發(fā)生錯誤時關閉連接，避免持續(xù)消耗資源。

4. 測試 SSE

運行 Spring Boot 應用，并訪問 /sse/stream，可以看到服務器每秒鐘向客戶端推送一次當前時間信息。

接口請求

header 里的 Content-Type 為 text/event-stream。

Content-Type

可以通過瀏覽器打開 http://localhost:8080/，在頁面中將會每秒鐘顯示一次服務器推送的數(shù)據流。這就驗證了 SSE 在 Spring Boot 3 中的實現(xiàn)。

瀏覽器展示

5. 優(yōu)化與擴展

5.1 增加隨機數(shù)據推送

為了模擬更真實的場景，可以增加一些隨機數(shù)據或實時數(shù)據更新。假設我們希望推送隨機的股票價格，我們可以這樣修改：

@GetMapping("/sse/stocks")
public Flux<ServerSentEvent<String>> streamStockPrices() {
    return Flux.interval(Duration.ofSeconds(1))
            .map(sequence -> ServerSentEvent.<String>builder()
                    .id(String.valueOf(sequence))
                    .event("stock-update")
                    .data("Stock price: $" + ThreadLocalRandom.current().nextInt(100, 200))
                    .build());
}

在這個例子中，每秒推送一次隨機的股票價格更新。

5.2 增加心跳檢測（Ping）

SSE 連接如果長時間沒有數(shù)據傳輸，可能會被中斷。為此，SSE 規(guī)范推薦發(fā)送 “ping” 消息來保持連接活躍?？梢酝ㄟ^ ServerSentEvent 的 comment() 來發(fā)送心跳信息：

@GetMapping("/sse/stream-with-ping")
public Flux<ServerSentEvent<String>> streamWithPing() {
    return Flux.interval(Duration.ofSeconds(1))
            .map(sequence -> {
                if (sequence % 5 == 0) {  // 每5秒發(fā)送一次心跳
                    return ServerSentEvent.<String>builder()
                            .comment("ping")
                            .build();
                } else {
                    return ServerSentEvent.<String>builder()
                            .data("Current time: " + LocalTime.now())
                            .build();
                }
            });
}

5.3 使用 MediaType.TEXT_EVENT_STREAM 響應

雖然 ServerSentEvent 是處理 SSE 的標準類，但你也可以直接返回 Flux<T>，Spring 會自動將其轉換為事件流。如果你想簡化代碼，可以這樣寫：

@GetMapping(value = "/sse/simple", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> simpleSse() {
    return Flux.interval(Duration.ofSeconds(1))
            .map(sequence -> "Current time: " + LocalTime.now());
}

這里直接返回 Flux<String>，Spring WebFlux 會自動推送數(shù)據。

6. SSE 與 WebSocket 的對比

SSE 和 WebSocket 都是實時通信的重要技術，但它們有不同的適用場景：

SSE：單向通信，服務器推送數(shù)據到客戶端，適合輕量級的通知、監(jiān)控、消息更新等場景。使用簡單，基于 HTTP。
WebSocket：雙向通信，適合復雜的交互場景，如實時聊天、在線游戲等。WebSocket 是基于 TCP 的全雙工連接，相對更復雜。

對于簡單的實時更新場景，如股票價格更新、推送通知等，SSE 更加輕量且易于實現(xiàn)。

7. 總結

Spring Boot 3 提供了簡單、強大的 SSE 實現(xiàn)，結合響應式編程的特性，使得我們可以輕松構建高效的服務器推送應用。在實際項目中，SSE 非常適合用于推送實時數(shù)據或監(jiān)控信息，尤其在需要輕量且可靠的單向通信時。通過 Spring WebFlux 和 Project Reactor，SSE 的實現(xiàn)可以以非阻塞的方式運行，極大提升了應用的并發(fā)處理能力。

希望這篇博客對你理解 Spring Boot 3 中的 SSE 服務端推送有所幫助，如果有任何問題或想法，歡迎討論！

到此這篇關于SpringBoot3中Spring WebFlux SSE服務器發(fā)送事件的實現(xiàn)步驟的文章就介紹到這了,更多相關SpringBoot SSE服務器發(fā)送事件內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: