SpringBoot+ResponseBodyEmitter實時異步流式推送的實現(xiàn)

更新時間：2025年11月18日 09:35:40 作者：皮皮林551

本文主要介紹了SpringBoot+ResponseBodyEmitter實時異步流式推送的實現(xiàn),文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧

ResponseBodyEmitter 的作用

相較于 SSE 技術，ResponseBodyEmitter 更加簡單易用。它主要用于處理異步的 HTTP 響應，其核心優(yōu)勢在于 允許逐步將數(shù)據(jù)發(fā)送到客戶端，而非一次性發(fā)送所有內(nèi)容。這一特性使得它在需要長時間處理或進行流式傳輸?shù)膱鼍爸斜憩F(xiàn)出色。需要注意的是，ResponseBodyEmitter 本質(zhì)上是一個接口。

使用場景

長輪詢：服務器在有數(shù)據(jù)時會立即響應客戶端請求，若暫無數(shù)據(jù)，則保持連接開放，等待數(shù)據(jù)到來。
服務器推送事件 (SSE)：服務器能夠持續(xù)不斷地向客戶端推送各類事件，實現(xiàn)實時交互。
流式傳輸：可逐步發(fā)送大量數(shù)據(jù)，像文件下載或者實時數(shù)據(jù)流傳輸?shù)葓鼍岸歼m用。
異步處理：在處理耗時任務時，能逐步返回處理結果，避免客戶端長時間等待，提升用戶體驗。

業(yè)務場景舉例

在實際業(yè)務中，ResponseBodyEmitter 有著廣泛的應用，比如進度條的實時更新、實時聊天功能、股票價格的實時更新、系統(tǒng)日志的流式輸出以及 AI 的流式響應等。

實時日志流實戰(zhàn)

接下來，我們通過一個簡單的實時日志流功能，來深入了解 ResponseBodyEmitter 的使用。假設我們有一個應用程序，需要實時查看服務器的日志，以便快速定位和解決問題。

創(chuàng)建控制器

首先，我們在 Spring Boot 應用中創(chuàng)建一個控制器，借助 ResponseBodyEmitter 實現(xiàn)實時日志流。

import org.springframework.http.MediaType;
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.ResponseBodyEmitter;

@RestController
@RequestMapping("/api/log")
publicclass LogController {
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public ResponseBodyEmitter streamLogs() {
        ResponseBodyEmitter emitter = new ResponseBodyEmitter();

        // 開啟異步線程處理數(shù)據(jù)并發(fā)送
        new Thread(() -> {
            try {
                while (true) {
                    String logEntry = getLatestLogEntry();
                    if (logEntry != null) {
                        emitter.send(logEntry);
                    }
                    // 每秒檢查一次日志更新
                    Thread.sleep(1000); 
                }
            } catch (Exception e) {
                // 出現(xiàn)異常時結束響應并傳遞錯誤信息
                emitter.completeWithError(e); 
            }
        }).start();

        return emitter;
    }

    private String getLatestLogEntry() {
        // 模擬從日志文件中獲取最新日志條目
        return"2025-02-12 12:00:00 - INFO: User logged in successfully.";
    }
}

運行效果

當我們啟動這個應用程序，并訪問 /api/log/stream 路徑時，就能看到一個實時更新的日志流。服務器會每秒向客戶端推送一條新的日志條目，客戶端會將其顯示在頁面上，效果如下：

ResponseBodyEmitter 的核心方法

send(Object data)：向客戶端發(fā)送數(shù)據(jù)，該方法可以多次調(diào)用，實現(xiàn)數(shù)據(jù)的逐步發(fā)送。
complete()：用于結束響應流，表示數(shù)據(jù)已經(jīng)全部發(fā)送完畢。
onTimeout(Runnable callback)：設置超時回調(diào)函數(shù)，當連接超時時，會執(zhí)行該回調(diào)。
onCompletion(Runnable callback)：設置完成回調(diào)函數(shù)，當數(shù)據(jù)發(fā)送完成后，會執(zhí)行該回調(diào)。

ResponseBodyEmitter 工作原理

異步數(shù)據(jù)生成與推送

在傳統(tǒng)的 HTTP 請求 - 響應模式中，服務器通常需要等待整個響應數(shù)據(jù)生成完成后，才會將其一次性發(fā)送給客戶端。而 ResponseBodyEmitter 打破了這種模式，它允許服務端在任務執(zhí)行過程中異步地生成響應數(shù)據(jù)。

當有部分數(shù)據(jù)準備好時，就可以立即調(diào)用 send() 方法將這些數(shù)據(jù)推送給客戶端，而無需等待整個任務完成。這就好比一場接力賽，每完成一段賽程（生成一部分數(shù)據(jù)），就馬上將接力棒（數(shù)據(jù)）傳遞給客戶端，大大提高了數(shù)據(jù)傳輸?shù)膶崟r性。

分塊傳輸機制

ResponseBodyEmitter 采用了 HTTP 的分塊編碼（Chunked Encoding）方式來傳輸數(shù)據(jù)。在傳統(tǒng)的 HTTP 響應中，通常需要在響應頭中明確指定 Content-Length，表示整個響應數(shù)據(jù)的長度。但在分塊傳輸中，服務器不會提前設置 Content-Length，而是將數(shù)據(jù)分成多個獨立的塊，每個塊都有自己的長度標識。

客戶端在接收到數(shù)據(jù)塊后，可以立即對其進行處理，而不必等待整個響應數(shù)據(jù)接收完畢。這種方式使得數(shù)據(jù)可以邊生成邊傳輸，減少了客戶端的等待時間，提高了用戶體驗。

連接生命周期管理

為了確保資源的合理使用，ResponseBodyEmitter 提供了對連接生命周期的有效管理。當所有數(shù)據(jù)都發(fā)送完畢后，需要調(diào)用 complete() 方法來明確告知客戶端響應結束，關閉連接。如果在數(shù)據(jù)傳輸過程中出現(xiàn)異常，可以調(diào)用 completeWithError() 方法，結束響應并向客戶端傳遞錯誤信息。

這樣可以避免連接長時間保持開放，造成資源浪費。

注意事項

客戶端支持：雖然大多數(shù)瀏覽器和 HTTP 客戶端庫都支持分塊傳輸，但某些老舊的客戶端可能存在兼容性問題。
超時設置：為避免長連接長時間占用資源，可以為 ResponseBodyEmitter 設置超時時間，示例代碼如下：

emitter.onTimeout(() -> emitter.complete());

線程安全：ResponseBodyEmitter 的 send() 方法是線程安全的，但在使用時需要注意控制任務線程的生命周期，避免出現(xiàn)資源泄漏。
連接關閉：務必確保在任務結束時調(diào)用 complete() 或 completeWithError() 方法，否則可能導致連接無法正常關閉，造成資源浪費。

與 Streaming 和 SSE 的對比

Streaming：直接通過 OutputStream 向客戶端寫入數(shù)據(jù)，靈活性較高，但需要手動處理流的關閉，增加了開發(fā)的復雜度。
Server-Sent Events (SSE)：基于 text/event-stream 協(xié)議，適用于服務端事件推送場景，但要求客戶端支持 SSE 協(xié)議。
ResponseBodyEmitter：通用性更強，適用于任何支持 HTTP 的客戶端，并且易于與 Spring 框架集成，是一種更為便捷的流式傳輸解決方案。

在處理類似 AI 這種響應式的流式輸出場景時，相較于 SSE，ResponseBodyEmitter 作為 Spring 提供的輕量級流式傳輸解決方案，在 HTTP 協(xié)議兼容性方面表現(xiàn)更優(yōu)。