Java調(diào)用SSE流式接口并流式返回給前端實(shí)現(xiàn)打字輸出效果

更新時(shí)間：2024年08月17日 15:27:39 作者：weixin_44680858

在Web開發(fā)中,有時(shí)我們需要將文件以流的形式返回給前端,下面這篇文章主要給大家介紹了關(guān)于Java調(diào)用SSE流式接口并流式返回給前端實(shí)現(xiàn)打字輸出效果的相關(guān)資料,需要的朋友可以參考下

1.SSE概述

1.1 什么是是SSE

Server-Sent Events (SSE)SSE是一種簡單的事件推送技術(shù)，它允許服務(wù)器異步地向客戶端發(fā)送更新，而無需客戶端顯式請求這些更新。這對于實(shí)時(shí)應(yīng)用程序非常有用，例如股票價(jià)格更新、消息通知等。SSE基于HTTP協(xié)議，使用一個(gè)持久的HTTP連接來維持客戶端和服務(wù)端之間的通信。

2.2 與長鏈接(Long Polling)的區(qū)別

Server-Sent Events (SSE) 和長鏈接(Long Polling)都是實(shí)現(xiàn)服務(wù)器向客戶端推送數(shù)據(jù)的技術(shù)，但它們之間存在一些關(guān)鍵區(qū)別。下面我將詳細(xì)解釋這兩種技術(shù)的不同之處：

長鏈接(Long Polling)

長鏈接是一種實(shí)現(xiàn)服務(wù)器推送數(shù)據(jù)到客戶端的技術(shù)，它基于HTTP請求/響應(yīng)模型。在這種模式下，客戶端發(fā)起一個(gè)HTTP請求，服務(wù)器在沒有數(shù)據(jù)可發(fā)送的情況下會(huì)保持連接打開，直到有數(shù)據(jù)可發(fā)送或者超時(shí)。一旦服務(wù)器有數(shù)據(jù)要發(fā)送，它就會(huì)響應(yīng)客戶端的請求，并關(guān)閉連接。客戶端接收到數(shù)據(jù)后立即重新發(fā)起一個(gè)新的請求，從而保持與服務(wù)器的“長鏈接”。

特點(diǎn)：

客戶端主動(dòng)發(fā)起請求：客戶端需要不斷地向服務(wù)器發(fā)起請求以獲取數(shù)據(jù)。
服務(wù)器被動(dòng)響應(yīng)：服務(wù)器只在客戶端請求時(shí)才發(fā)送數(shù)據(jù)。
連接短暫：雖然每個(gè)連接可能會(huì)持續(xù)一段時(shí)間，但每次請求結(jié)束后連接會(huì)被關(guān)閉。
實(shí)現(xiàn)簡單：易于用現(xiàn)有HTTP技術(shù)實(shí)現(xiàn)。
兼容性好：幾乎所有瀏覽器都支持HTTP請求/響應(yīng)模型。

Server-Sent Events (SSE)

Server-Sent Events 是一種更為現(xiàn)代的技術(shù)，用于實(shí)現(xiàn)服務(wù)器向客戶端的單向數(shù)據(jù)推送。SSE基于HTTP協(xié)議，但使用了一個(gè)持久的HTTP連接來維持客戶端和服務(wù)端之間的通信。服務(wù)器可以主動(dòng)向客戶端發(fā)送數(shù)據(jù)，而不需要等待客戶端的請求。

特點(diǎn)：

服務(wù)器主動(dòng)推送：服務(wù)器可以主動(dòng)向客戶端發(fā)送數(shù)據(jù)，而不需要客戶端發(fā)起請求。
持久連接：客戶端和服務(wù)端之間建立了一個(gè)持久的連接，直到客戶端或服務(wù)器關(guān)閉該連接。
格式特定：SSE使用特定的格式來發(fā)送數(shù)據(jù)，包括data:字段和空行作為分隔符。
資源效率高：由于連接是持久的，因此減少了建立連接的開銷。
實(shí)現(xiàn)復(fù)雜度適中：雖然比長鏈接稍微復(fù)雜，但現(xiàn)代瀏覽器和服務(wù)器框架提供了良好的支持。

比較

實(shí)時(shí)性：SSE提供更好的實(shí)時(shí)性，因?yàn)樗恍枰蛻舳瞬粩喟l(fā)起請求。
性能：SSE在性能上通常優(yōu)于長鏈接，因?yàn)樗苊饬酥貜?fù)建立連接的開銷。
實(shí)現(xiàn)復(fù)雜度：SSE需要客戶端和服務(wù)端雙方的支持，而長鏈接可以更容易地在現(xiàn)有的HTTP基礎(chǔ)設(shè)施上實(shí)現(xiàn)。
兼容性：SSE在現(xiàn)代瀏覽器中得到了廣泛支持，但對于一些舊版瀏覽器可能不適用；長鏈接則具有更好的向后兼容性。

總結(jié)

選擇哪種技術(shù)取決于你的具體需求。如果你的應(yīng)用需要較低延遲的數(shù)據(jù)推送，并且可以依賴現(xiàn)代瀏覽器和服務(wù)器環(huán)境，那么SSE是一個(gè)不錯(cuò)的選擇。如果你需要更廣泛的瀏覽器兼容性，并且對實(shí)時(shí)性要求不是特別高，那么長鏈接可能更適合你。

2.通過okhttp調(diào)用SSE流式接口并流式返回給前端

環(huán)境要求

Spring Framework 5.0
Jdk1.8

使用okhttp相關(guān)依賴

<dependency>
    <groupId>com.squareup.okhttp3</groupId>
    <artifactId>okhttp</artifactId>
    <version>4.2.0</version>
</dependency>
<dependency>
    <groupId>com.squareup.okhttp3</groupId>
    <artifactId>okhttp-sse</artifactId>
    <version>4.2.0</version>
</dependency>

示例

@GetMapping(value = "/test1", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter SseTest1() {
        SseEmitter sseEmitter = new SseEmitter();
        String prompt = "";
        String url = "";
        FormBody formBody = new FormBody.Builder().add("prompt", prompt).build();
        Request request = new Request.Builder().url(url).post(formBody).build();
        // 使用EventSourceListener處理來自服務(wù)器的SSE事件
        EventSourceListener listener = new EventSourceListener() {
            @Override
            public void onOpen(@NotNull EventSource eventSource, @NotNull Response response) {
                log.info("Connection opened.");
            }
            @Override
            public void onClosed(@NotNull EventSource eventSource) {
                log.info("Connection closed.");
                sseEmitter.complete();
            }
            @Override
            public void onEvent(@NotNull EventSource eventSource, @Nullable String id, @Nullable String type, @NotNull String data) {
                try {
                    JSONObject jsonObject = JSONUtil.parseObj(data);
                    String event = jsonObject.getStr("event");
                    if ("message".equals(event)) {
                        sseEmitter.send(jsonObject.getStr("answer"));
                    }
                } catch (Exception e) {
                    log.error("推送數(shù)據(jù)失敗", e);
                }
            }
            @Override
            public void onFailure(@NotNull EventSource eventSource, @Nullable Throwable t, @Nullable Response response) {
                log.error("Connection failed.", t);
                sseEmitter.completeWithError(t);
            }
        };
        OkHttpClient client = new OkHttpClient.Builder().connectTimeout(10, TimeUnit.SECONDS).writeTimeout(50, TimeUnit.SECONDS).readTimeout(10, TimeUnit.MINUTES).build();
        EventSource.Factory factory = EventSources.createFactory(client);
        factory.newEventSource(request, listener);
        return sseEmitter;
    }

注意

該接口需為Get請求，ContentType為 text/event-stream
SseEmitter 是Spring Framework 5.0引入的一個(gè)新特性，用于簡化Server-Sent Events (SSE) 的實(shí)現(xiàn)。它提供了一種簡單的方式來發(fā)送事件數(shù)據(jù)到客戶端，特別適用于構(gòu)建實(shí)時(shí)數(shù)據(jù)推送的應(yīng)用程序。

3. 如果Spring Framework 低于5.0，可使用Servlet 3.0進(jìn)行流式返回

使用AsyncContext：Servlet 3.0 引入了異步支持，允許Servlet在不同的線程中處理請求。你可以使用AsyncContext來啟動(dòng)一個(gè)異步線程，在該線程中發(fā)送SSE事件。

配置async-supported使用AsyncContext前需配置async-supported

async-supported元素用于指定Servlet是否支持異步處理。這個(gè)配置通常是在部署描述符 web.xml 文件中進(jìn)行設(shè)置的。

配置示例

<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
         version="3.1">

    <servlet>
        <servlet-name>MyServlet</servlet-name>
        <servlet-class>com.example.MyServlet</servlet-class>
        <async-supported>true</async-supported>
    </servlet>

    <servlet-mapping>
        <servlet-name>MyServlet</servlet-name>
        <url-pattern>/myServlet</url-pattern>
    </servlet-mapping>

</web-app>

后端代碼示例

@GetMapping("/test3")
    public void SseTest3(HttpServletRequest req, HttpServletResponse resp) {
        resp.setContentType("text/event-stream");
        resp.setCharacterEncoding("UTF-8");
        resp.setHeader("Cache-Control", "no-cache");
        resp.setHeader("Connection", "keep-alive");

        try {
            //
            AsyncContext asyncContext = req.startAsync(req, resp);
            asyncContext.setTimeout(10 * 60 * 1000);
            PrintWriter writer = asyncContext.getResponse().getWriter();

            String prompt = "";
            String url = "";
            FormBody formBody = new FormBody.Builder().add("prompt", prompt).build();
            Request request = new Request.Builder().url(url).post(formBody).build();
            // 使用EventSourceListener處理來自服務(wù)器的SSE事件
            EventSourceListener listener = new EventSourceListener() {
                @Override
                public void onOpen(@NotNull EventSource eventSource, @NotNull Response response) {
                    log.info("Connection opened.");
                }

                @Override
                public void onClosed(@NotNull EventSource eventSource) {
                    log.info("Connection closed.");
                    writer.write("data: __stop__\n\n");
                    writer.flush();
                    asyncContext.complete();
                }

                @Override
                public void onEvent(@NotNull EventSource eventSource, @Nullable String id, @Nullable String type, @NotNull String data) {
                    try {
                        JSONObject jsonObject = JSONUtil.parseObj(data);
                        String event = jsonObject.getStr("event");
                        if ("message".equals(event)) {
                            String answer = jsonObject.getStr("answer");
                            log.info("message: {}", answer);
                            writer.write("data: " + answer + "\n\n");
                            writer.flush();
                        }
                    } catch (Exception e) {
                        log.error("推送數(shù)據(jù)失敗", e);
                    }
                }

                @Override
                public void onFailure(@NotNull EventSource eventSource, @Nullable Throwable t, @Nullable Response response) {
                    log.error("Connection failed.", t);
                    asyncContext.complete();
                }
            };
            OkHttpClient client = new OkHttpClient.Builder().connectTimeout(10, TimeUnit.SECONDS).writeTimeout(50, TimeUnit.SECONDS).readTimeout(10, TimeUnit.MINUTES).build();
            EventSource.Factory factory = EventSources.createFactory(client);
            factory.newEventSource(request, listener);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {

        }
    }

注意返回?cái)?shù)據(jù)格式:

// 以data: 開頭  /n/n結(jié)束
"data: xxxxx /n/n"

4. 前端調(diào)用SSE接口

方式1 使用JavaScript的 EventSource API

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SSE Example</title>
</head>
<body>
    <div id="events"></div>
    <script>
        const source = new EventSource('/sse');

        source.onmessage = function(event) {
            const data = JSON.parse(event.data);
            // 約定一個(gè)結(jié)束標(biāo)識
            if(data == '__stop__') {
            	source.close()
            	return
            }
            document.getElementById('events').innerHTML += `<p>${data.message}</p>`;
        };

        source.onerror = function(error) {
            console.error('Error occurred:', error);
            source.close();
        };
    </script>
</body>
</html>

注意

后端返回需返回完整消息的對象（包括換行符），例：{“data”: “哈哈哈/n/n”}，如果后端將data取出，則會(huì)導(dǎo)致?lián)Q行符丟失！
EventSource 只支持Get請求，如果請求參數(shù)過長會(huì)導(dǎo)致調(diào)用失??！

方式2 使用 fetchEventSource 插件

安裝插件

npm install --save @microsoft/fetch-event-source

簡單示例

// 導(dǎo)入依賴
import { fetchEventSource } from '@microsoft/fetch-event-source';
 
send() {
	const vm = this;
	const ctrlAbout = new AbortController();
	const { signal } = ctrlAbout;
	fetchEventSource(Url, {
	  method: 'POST',
	  headers: {
	    "Content-Type": 'application/json',
	    "Accept": 'text/event-stream'
	  },
	  body: JSON.stringify(data),
	  signal: ctrl.signal, // AbortSignal
	  onmessage(event) {
	     console.info(event.data);
	     // 在這里操作流式數(shù)據(jù)
	     const message = JSON.parse(event.data)
	     vm.content += message.data
	  },
	  onclose(e) {
	     // 關(guān)閉流
	     // 中斷流式返回
	     ctrl.abort()
	  }
	  onerror(error) {
	    // 返回流報(bào)錯(cuò)
		console.info(error);
		// 中斷流式返回
		ctrl.abort()
		throw err // 直接拋出錯(cuò)誤，避免反復(fù)調(diào)用
	  }
	})
}

注意

傳參時(shí)需注意參數(shù)類型為json字符串

5. 使用原生的http調(diào)用SSE流式接口

示例

@GetMapping("/test2")
    public void SseTest2() {
        String urlAddr = "";
        BufferedReader reader = null;
        try {
            URL url = new URL(urlAddr);
            // 建立鏈接
            HttpURLConnection connection = (HttpURLConnection) url.openConnection();
            connection.setRequestMethod("POST");
            connection.setRequestProperty("Accept", "text/event-stream");
            connection.setRequestProperty("Content-type", "application/json; charset=UTF-8");
            connection.setRequestProperty("Cache-Control", "no-cache");
            connection.setRequestProperty("Connection", "keep-alive");
            // 允許輸入和輸出
            connection.setDoInput(true);
            connection.setDoOutput(true);
            // 設(shè)置超時(shí)為0，表示無限制
            connection.setConnectTimeout(0);
            connection.setReadTimeout(0);
            // 傳參
            String params = "prompt=哈哈哈哈";
            // 寫入POST數(shù)據(jù)
            DataOutputStream out = new DataOutputStream(connection.getOutputStream());
            out.write(params.getBytes(StandardCharsets.UTF_8));
            out.flush();
            out.close();

            // 讀取SSE事件
            reader = new BufferedReader(new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8));
            StringBuilder eventBuilder = new StringBuilder();
            String line;
            while ((line = reader.readLine()) != null) {
                System.out.println(line);
            }
            reader.close();
            // 斷開鏈接
            connection.disconnect();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            IoUtil.close(reader);
        }
    }

總結(jié)

到此這篇關(guān)于Java調(diào)用SSE流式接口并流式返回給前端實(shí)現(xiàn)打字輸出效果的文章就介紹到這了,更多相關(guān)Java調(diào)用SSE流式接口并返回前端內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: