java后端http接口流式輸出到前端的方法示例

更新時間：2025年09月19日 10:37:39 作者：abcnull

這篇文章主要介紹了java后端http接口流式輸出到前端的相關(guān)資料,對比SseEmitter與WebFlux兩種實現(xiàn)方式,通過實例代碼詳細(xì)分析了其適用場景及優(yōu)劣,需要的朋友可以參考下

前置

解釋：

Server-Sent Events：服務(wù)器發(fā)送事件，是一種基于 HTTP 的輕量級協(xié)議，允許服務(wù)器主動向客戶端推送文本數(shù)據(jù)（如 JSON、純文本等）

特點：

單向通信：僅服務(wù)器 → 客戶端方向
基于 HTTP/HTTPS：無需特殊協(xié)議
自動重連：瀏覽器內(nèi)置支持?jǐn)嗑€重連
簡單易用：前端直接使用 EventSource API

流程：

前端 (Vue) → 發(fā)起SSE請求 → Spring Boot (Controller)
↓ ↓
(EventSource) ← 流式數(shù)據(jù) ← WebClient → 大模型API (流式HTTP)

后端流式輸出

使用 springboot SseEmitter

傳統(tǒng)項目小范圍流式推送 → SseEmitter（改動成本低）

輕量級 SSE 推送、兼容舊 MVC 項目

每個請求占用一個線程，高并發(fā)時資源消耗大

@RestController
public class StreamingController {
    @GetMapping("/stream")
    public SseEmitter streamText() {
        SseEmitter emitter = new SseEmitter(60_000L); // 設(shè)置超時時間（毫秒）

        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(() -> {
            try {
                for (int i = 1; i <= 10; i++) {
                    String data = "這是第 " + i + " 段內(nèi)容\n";
                    emitter.send(SseEmitter.event().name("message").data(data));
                    Thread.sleep(1000); // 模擬延遲
                }
                emitter.complete(); // 結(jié)束流
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e);
            } finally {
                executor.shutdown();
            }
        });

        return emitter;
    }
}

代碼內(nèi)部直接承接大模型 api 調(diào)用，然后流式輸出到前端

@RestController
public class ModelStreamController {
    // 創(chuàng)建線程池
    private final ExecutorService executor = Executors.newCachedThreadPool();
    @GetMapping("/model-stream")
    public SseEmitter streamModel() {
        // 創(chuàng)建 SSE 發(fā)射器（60秒超時）
        SseEmitter emitter = new SseEmitter(60_000L);
        
        // 提交任務(wù)到線程池
        executor.execute(() -> {
            try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
                // 1. 創(chuàng)建到大模型API的請求
                HttpGet request = new HttpGet("https://api.open-model.com/stream");
                // 2. 執(zhí)行請求
                HttpResponse response = httpClient.execute(request);
                // 3. 檢查響應(yīng)狀態(tài)
                if (response.getStatusLine().getStatusCode() != 200) {
                    throw new RuntimeException("API error: " + 
                            response.getStatusLine().getStatusCode());
                }
                
                // 4. 獲取響應(yīng)流
                try (InputStream is = response.getEntity().getContent();
                     BufferedReader reader = new BufferedReader(new InputStreamReader(is))) {
                    String line;
                    // 5. 逐行讀取并轉(zhuǎn)發(fā)
                    while ((line = reader.readLine()) != null) {
                        // 6. 發(fā)送給前端
                        emitter.send(line);
                    }
                    
                    // 7. 完成后關(guān)閉SSE
                    emitter.complete();
                }
            } catch (Exception e) {
                // 8. 錯誤處理
                emitter.completeWithError(e);
            }
        });
        return emitter;
    }
}

使用 springboot WebFlux

新建高并發(fā)/代理外部流式 API → Flux + WebFlux（性能與擴展性更優(yōu)）

響應(yīng)式編程（Reactive Streams）

每個請求占用一個線程，高并發(fā)時資源消耗大

全鏈路非阻塞，適合代理外部流式 API

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId> <!-- 非阻塞IO -->
</dependency>

下面例子是轉(zhuǎn)發(fā)大模型的響應(yīng)結(jié)果

@RestController
public class WebFluxController {
    @GetMapping(value = "/flux-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> stream() {
        // 直接轉(zhuǎn)發(fā)大模型的流式響應(yīng)（非阻塞）
        WebClient webClient = WebClient.create("https://api.open-model.com/stream");
        return webClient.get()
            .retrieve()
            .bodyToFlux(String.class)
            .map(data -> "data: " + data + "\n\n") // 封裝為 SSE 格式
            .delayElements(Duration.ofMillis(100)); // 非阻塞延遲
    }
}

使用 servlet

protected void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException {
    // 設(shè)置響應(yīng)頭，聲明內(nèi)容類型為 SSE
    response.setContentType("text/event-stream");
    response.setCharacterEncoding("UTF-8");
    response.setHeader("Cache-Control", "no-cache");
    response.setHeader("Connection", "keep-alive");

    OutputStream outputStream = response.getOutputStream();

    try {
        // 模擬流式傳輸數(shù)據(jù)
        for (int i = 0; i < 5; i++) {
            String data = "event: message\ndata: {" + i + "}\n\n";
            outputStream.write(data.getBytes()); // 寫入
            outputStream.flush(); // 不斷的發(fā)送 flush
            Thread.sleep(1000); // 模擬延遲
        }

        // 發(fā)送自定義結(jié)束事件
        String endEvent = "data: [DONE]\n\n";
        outputStream.write(endEvent.getBytes());
        outputStream.flush();

        // 關(guān)閉流，通知前端結(jié)束
        outputStream.close();
    } catch (Exception e) {
        e.printStackTrace();
        outputStream.close();
    }
}

如果是你的服務(wù)接入了大模型 api，大模型本身做的就是一個流式輸出呢？承接后直接輸出到前端

@WebServlet("/proxy-stream")
public class StreamingProxyServlet extends HttpServlet {
    @Override
    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        // 設(shè)置響應(yīng)頭
        response.setContentType("text/event-stream");
        response.setCharacterEncoding("UTF-8");
        response.setHeader("Cache-Control", "no-cache");
        response.setHeader("Connection", "keep-alive");

        // 構(gòu)造 JSON 請求體
        String jsonBody = "{\"param1\": \"value1\", \"param2\": \"value2\"}";

        // 創(chuàng)建 HttpClient 實例
        try (CloseableHttpClient httpClient = HttpClients.createDefault()) {
            HttpPost httpPost = new HttpPost("https://third-party.com/stream"); // 替換為你的流式接口地址
            // 設(shè)置請求頭
            httpPost.setHeader("Content-Type", "application/json; charset=UTF-8");
            // 設(shè)置請求體
            httpPost.setEntity(new StringEntity(jsonBody, StandardCharsets.UTF_8));
            // 執(zhí)行請求
            try (CloseableHttpResponse backendResponse = httpClient.execute(httpPost)) {
                // 檢查后端響應(yīng)狀態(tài)碼
                int statusCode = backendResponse.getStatusLine().getStatusCode();
                if (statusCode != 200) {
                    response.sendError(HttpServletResponse.SC_BAD_GATEWAY, "后端接口返回狀態(tài)碼: " + statusCode);
                    return;
                }

                // 獲取后端流式接口的輸入流
                InputStream backendStream = backendResponse.getEntity().getContent();
                // 獲取前端響應(yīng)的輸出流
                OutputStream frontendStream = response.getOutputStream();
                byte[] buffer = new byte[4096];
                int bytesRead;

                // 邊讀邊寫，實時透傳
                while ((bytesRead = backendStream.read(buffer)) != -1) {
                    frontendStream.write(buffer, 0, bytesRead);
                    frontendStream.flush(); // 必須立即刷新，確保數(shù)據(jù)實時到達(dá)前端
                }

                // 結(jié)束流式連接
                frontendStream.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
            response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "流式傳輸失敗: " + e.getMessage());
        }
    }
}