SpringBoot + ResponseBodyEmitter 實(shí)時(shí)異步流式推送，優(yōu)雅！

作者：不才陳某 2025-03-10 00:00:11

ResponseBodyEmitter? 是 Spring 框架提供的輕量級(jí)流式傳輸解決方案，它能夠顯著提升高并發(fā)和實(shí)時(shí)性場(chǎng)景下的用戶體驗(yàn)。通過 ResponseBodyEmitter，我們可以輕松實(shí)現(xiàn)服務(wù)器向客戶端的實(shí)時(shí)數(shù)據(jù)推送。

ChatGPT 的火爆，讓流式輸出技術(shù)迅速走進(jìn)大眾視野。在那段時(shí)間里，許多熱愛鉆研技術(shù)的小伙伴紛紛開始學(xué)習(xí)和實(shí)踐 SSE 異步處理。

我當(dāng)時(shí)也寫過相關(guān)文章，今天，咱們換一種更為簡(jiǎn)便的方式來實(shí)現(xiàn)流式輸出，那就是 ResponseBodyEmitter。

其實(shí)，ResponseBodyEmitter 并非新技術(shù)，早在 Spring Framework 4.2 版本就已被引入。直到最近，我們?cè)陂_發(fā)一個(gè)滾動(dòng)日志輸出功能時(shí)，才深入了解到它的強(qiáng)大之處。

ResponseBodyEmitter 的作用

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

使用場(chǎng)景

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

業(yè)務(wù)場(chǎng)景舉例

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

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

接下來，我們通過一個(gè)簡(jiǎn)單的實(shí)時(shí)日志流功能，來深入了解 ResponseBodyEmitter 的使用。假設(shè)我們有一個(gè)應(yīng)用程序，需要實(shí)時(shí)查看服務(wù)器的日志，以便快速定位和解決問題。

創(chuàng)建控制器

首先，我們?cè)?Spring Boot 應(yīng)用中創(chuàng)建一個(gè)控制器，借助 ResponseBodyEmitter 實(shí)現(xiàn)實(shí)時(shí)日志流。

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)異常時(shí)結(jié)束響應(yīng)并傳遞錯(cuò)誤信息
                emitter.completeWithError(e); 
            }
        }).start();

        return emitter;
    }

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

運(yùn)行效果

當(dāng)我們啟動(dòng)這個(gè)應(yīng)用程序，并訪問 /api/log/stream 路徑時(shí)，就能看到一個(gè)實(shí)時(shí)更新的日志流。服務(wù)器會(huì)每秒向客戶端推送一條新的日志條目，客戶端會(huì)將其顯示在頁面上，效果如下：

運(yùn)行效果

ResponseBodyEmitter 的核心方法

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

ResponseBodyEmitter 工作原理

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

在傳統(tǒng)的 HTTP 請(qǐng)求 - 響應(yīng)模式中，服務(wù)器通常需要等待整個(gè)響應(yīng)數(shù)據(jù)生成完成后，才會(huì)將其一次性發(fā)送給客戶端。關(guān)注公眾號(hào)：碼猿技術(shù)專欄，回復(fù)關(guān)鍵詞：1111 獲取阿里內(nèi)部Java性能調(diào)優(yōu)手冊(cè)！而 ResponseBodyEmitter 打破了這種模式，它允許服務(wù)端在任務(wù)執(zhí)行過程中異步地生成響應(yīng)數(shù)據(jù)。

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

分塊傳輸機(jī)制

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

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

連接生命周期管理

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

這樣可以避免連接長(zhǎng)時(shí)間保持開放，造成資源浪費(fèi)。

注意事項(xiàng)

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

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

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

與 Streaming 和 SSE 的對(duì)比

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

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

小結(jié)

ResponseBodyEmitter 是 Spring 框架提供的輕量級(jí)流式傳輸解決方案，它能夠顯著提升高并發(fā)和實(shí)時(shí)性場(chǎng)景下的用戶體驗(yàn)。通過 ResponseBodyEmitter，我們可以輕松實(shí)現(xiàn)服務(wù)器向客戶端的實(shí)時(shí)數(shù)據(jù)推送。

無論是進(jìn)度條的實(shí)時(shí)更新、實(shí)時(shí)聊天、股票價(jià)格的實(shí)時(shí)監(jiān)控還是系統(tǒng)日志的流式輸出，ResponseBodyEmitter 都能幫助我們構(gòu)建更加動(dòng)態(tài)和互動(dòng)的應(yīng)用程序。

責(zé)任編輯：武曉燕來源：碼猿技術(shù)專欄