SpringBoot3中Spring?WebFlux?SSE服務(wù)器發(fā)送事件的實現(xiàn)步驟
ChatGPT 剛出的時候，讓大伙很好奇的是它是如何實現(xiàn)的逐字輸出的？答案就是 SSE （服務(wù)器發(fā)送事件）。隨著實時數(shù)據(jù)和響應(yīng)式編程的需求不斷增加，服務(wù)器發(fā)送事件（Server-Sent Events，簡稱 SSE）在現(xiàn)代 Web 應(yīng)用程序中越來越受歡迎。SSE 提供了一種輕量級的服務(wù)器推送數(shù)據(jù)給客戶端的方式，適合用于監(jiān)控、實時通知、股票價格更新等場景。
在 Spring Boot 3 中，結(jié)合響應(yīng)式編程的理念，SSE 的實現(xiàn)變得更加簡潔和高效。本文將詳細介紹如何使用 Spring Boot 3 來實現(xiàn) SSE 服務(wù)端推送，并討論響應(yīng)式編程在此過程中的重要性和優(yōu)勢。
1. 什么是 SSE？
服務(wù)器發(fā)送事件（SSE） 是一種從服務(wù)器向客戶端推送數(shù)據(jù)的技術(shù)，屬于 HTML5 的一部分。與傳統(tǒng)的 HTTP 請求-響應(yīng)模型不同，SSE 是單向的，服務(wù)器可以持續(xù)不斷地向客戶端發(fā)送數(shù)據(jù)，而客戶端通過一次長連接持續(xù)接收這些更新。
相比 WebSocket，SSE 有以下特點：
- 單向通信：SSE 僅允許服務(wù)器向客戶端推送數(shù)據(jù)，客戶端無法向服務(wù)器發(fā)送數(shù)據(jù)。
- 基于 HTTP 協(xié)議：SSE 是建立在 HTTP 協(xié)議之上的，瀏覽器原生支持，不需要額外的協(xié)議處理。
- 自動重連：SSE 支持自動重連，當連接意外斷開時，客戶端會自動嘗試重新連接服務(wù)器。
2. Spring Boot 3 響應(yīng)式編程與 SSE
Spring Boot 3 提供了對響應(yīng)式編程的全面支持，基于 Project Reactor 實現(xiàn)異步、非阻塞的流式數(shù)據(jù)處理。而響應(yīng)式編程非常適合實現(xiàn) SSE，因為它允許我們以非阻塞的方式持續(xù)推送數(shù)據(jù)，而不會阻塞服務(wù)器的資源。
Spring WebFlux 是 Spring Boot 3 中用于構(gòu)建響應(yīng)式應(yīng)用的核心框架，它可以無縫集成 SSE，為我們提供簡單高效的服務(wù)器推送功能。
為什么選擇響應(yīng)式編程實現(xiàn) SSE？
傳統(tǒng)的阻塞式編程在處理長連接（如 SSE）時可能會占用大量服務(wù)器資源。響應(yīng)式編程通過非阻塞 I/O 操作，不僅可以高效處理長時間的連接，還能在有新數(shù)據(jù)時立即推送給客戶端。響應(yīng)式流（如 Flux）天然適合于這種流式數(shù)據(jù)推送場景。
3. 實現(xiàn) SSE 的基本步驟
我們將通過以下步驟實現(xiàn)一個簡單的 SSE 服務(wù)端推送應(yīng)用：
- 創(chuàng)建 Spring Boot 項目并引入 WebFlux 依賴。
- 實現(xiàn)服務(wù)端推送 SSE 事件流。
- 編寫客戶端接收 SSE 數(shù)據(jù)。
- 測試與優(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)服務(wù)端推送SSE 事件流
在 Spring WebFlux 中，SSE 通過返回 Flux<ServerSentEvent<T>> 這種響應(yīng)流來實現(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ā)出事件的響應(yīng)式流。
ServerSentEvent.builder()：構(gòu)建 ServerSentEvent 對象，它可以包含 id、event 和 data 等信息，符合 SSE 規(guī)范。
map()：將流中的每個事件映射為 ServerSentEvent，并附帶當前的時間信息。
3.3 客戶端接收 SSE 數(shù)據(jù)
客戶端可以使用 JavaScript 原生的 EventSource API 來接收服務(wù)器發(fā)送的 SSE 數(shù)據(jù)流。
示例 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)有內(nèi)容
messagesElement.innerHTML = "";
// 添加新內(nèi)容
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 是瀏覽器提供的一個用于和服務(wù)器建立連接，接收服務(wù)器發(fā)送事件的接口。在客戶端發(fā)起與服務(wù)器的 SSE 長連接。服務(wù)器通過/sse/stream推送事件。onmessage：處理服務(wù)器發(fā)送的消息，并將消息顯示在頁面上。onerror：當連接發(fā)生錯誤時關(guān)閉連接，避免持續(xù)消耗資源。
4. 測試 SSE
運行 Spring Boot 應(yīng)用，并訪問 /sse/stream，可以看到服務(wù)器每秒鐘向客戶端推送一次當前時間信息。
header 里的 Content-Type 為 text/event-stream。
可以通過瀏覽器打開 http://localhost:8080/，在頁面中將會每秒鐘顯示一次服務(wù)器推送的數(shù)據(jù)流。這就驗證了 SSE 在 Spring Boot 3 中的實現(xiàn)。
5. 優(yōu)化與擴展
5.1 增加隨機數(shù)據(jù)推送
為了模擬更真實的場景，可以增加一些隨機數(shù)據(jù)或?qū)崟r數(shù)據(jù)更新。假設(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ù)據(jù)傳輸，可能會被中斷。為此，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 響應(yīng)
雖然 ServerSentEvent 是處理 SSE 的標準類，但你也可以直接返回 Flux<T>，Spring 會自動將其轉(zhuǎn)換為事件流。如果你想簡化代碼，可以這樣寫：
@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ù)據(jù)。
6. SSE 與 WebSocket 的對比
SSE 和 WebSocket 都是實時通信的重要技術(shù)，但它們有不同的適用場景：
- SSE：單向通信，服務(wù)器推送數(shù)據(jù)到客戶端，適合輕量級的通知、監(jiān)控、消息更新等場景。使用簡單，基于 HTTP。
- WebSocket：雙向通信，適合復(fù)雜的交互場景，如實時聊天、在線游戲等。WebSocket 是基于 TCP 的全雙工連接，相對更復(fù)雜。
對于簡單的實時更新場景，如股票價格更新、推送通知等，SSE 更加輕量且易于實現(xiàn)。
7. 總結(jié)
Spring Boot 3 提供了簡單、強大的 SSE 實現(xiàn)，結(jié)合響應(yīng)式編程的特性，使得我們可以輕松構(gòu)建高效的服務(wù)器推送應(yīng)用。在實際項目中，SSE 非常適合用于推送實時數(shù)據(jù)或監(jiān)控信息，尤其在需要輕量且可靠的單向通信時。通過 Spring WebFlux 和 Project Reactor，SSE 的實現(xiàn)可以以非阻塞的方式運行，極大提升了應(yīng)用的并發(fā)處理能力。
希望這篇博客對你理解 Spring Boot 3 中的 SSE 服務(wù)端推送有所幫助，如果有任何問題或想法，歡迎討論！
到此這篇關(guān)于SpringBoot3中Spring WebFlux SSE服務(wù)器發(fā)送事件的實現(xiàn)步驟的文章就介紹到這了,更多相關(guān)SpringBoot SSE服務(wù)器發(fā)送事件內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
在Spring框架中,循環(huán)依賴是指兩個或多個Bean相互依賴,這導(dǎo)致在Bean的創(chuàng)建過程中出現(xiàn)依賴死鎖,為了解決這一問題,Spring引入了三級緩存機制,包括singletonObjects、earlySingletonObjects和singletonFactories2024-09-09
這篇文章主要介紹了MyBatis sql中test如何判斷Boolean，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-01-01
spring事件監(jiān)聽機制離不開容器IOC特性提供的支持，比如容器會自動創(chuàng)建事件發(fā)布器，自動識別用戶注冊的監(jiān)聽器并進行管理，在特定的事件發(fā)布后會找到對應(yīng)的事件監(jiān)聽器并對其監(jiān)聽方法進行回調(diào)，這篇文章主要介紹了Spring事件監(jiān)聽源碼解析,需要的朋友可以參考下2023-08-08
多態(tài)就是指程序中定義的引用變量所指向的具體類型和通過該引用變量發(fā)出的方法調(diào)用在編程時并不確定，而是在程序運行期間才確定，即一個引用變量到底會指向哪個類的實例對象，該引用變量發(fā)出的方法調(diào)用到底是哪個類中實現(xiàn)的方法，必須在由程序運行期間才能決定2022-03-03
Java實戰(zhàn)之課程信息管理系統(tǒng)的實現(xiàn)
這篇文章主要介紹了如何利用Java實現(xiàn)課程信息管理系統(tǒng)，文中采用到的技術(shù)有：Springboot、SpringMVC、MyBatis、FreeMarker等，感興趣的可以了解一下2022-04-04
Maven是java中非常有用和常用的構(gòu)建工具，基本上現(xiàn)在大型的java項目都是Maven和gradle的天下了。本文將介紹Apache Maven ToolChains的使用。2021-06-06
Mybatis動態(tài)SQL之if、choose、where、set、trim、foreach標記實例詳解
動態(tài)SQL就是動態(tài)的生成SQL。接下來通過本文給大家介紹Mybatis動態(tài)SQL之if、choose、where、set、trim、foreach標記實例詳解的相關(guān)知識，感興趣的朋友一起看看吧2016-09-09
這篇文章主要介紹了Java方法的返回值及注意事項,本文給大家介紹的非常詳細，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-04-04
Java是一門天然的面向?qū)ο蟮恼Z言。而所有我們手動創(chuàng)造出來的類，都繼承于同一個類，即Object類。本文將通過示例為大家詳細介紹一下Java中的Object類，需要的可以參考一下2022-03-03
Java日期毫秒值和常見日期時間格式相互轉(zhuǎn)換方法
這篇文章主要給大家介紹了關(guān)于Java日期毫秒值和常見日期時間格式相互轉(zhuǎn)換的相關(guān)資料,在Java的日常開發(fā)中,會隨時遇到需要對時間處理的情況,文中給出了詳細的示例代碼,需要的朋友可以參考下2023-07-07