Java调用SSE流式接口,并流式返回给前端实现打字输出效果 ...

打印 上一主题 下一主题

主题 873|帖子 873|积分 2619

目次





    • 1.SSE概述


      • 1.1 什么是是SSE
      • 2.2 与长链接(Long Polling)的区别


        • 长链接(Long Polling)
        • Server-Sent Events (SSE)

      • 比力
      • 总结

    • 2.通过okhttp调用SSE流式接口并流式返回给前端




        • 环境要求
        • 使用okhttp相关依靠
        • 示例


    • 3. 如果Spring Framework 低于5.0,可使用Servlet 3.0举行流式返回
    • 4. 前端调用SSE接口


      • 方式1 使用JavaScript的 EventSource API
      • 方式2 使用 fetchEventSource 插件

    • 5. 使用原生的http调用SSE流式接口

1.SSE概述

1.1 什么是是SSE

Server-Sent Events (SSE)
SSE是一种简单的事件推送技能,它答应服务器异步地向客户端发送更新,而无需客户端显式请求这些更新。这对于实时应用步调非常有效,比方股票价格更新、消息通知等。SSE基于HTTP协议,使用一个持久的HTTP连接来维持客户端和服务端之间的通信。
2.2 与长链接(Long Polling)的区别

Server-Sent Events (SSE) 和长链接(Long Polling)都是实现服务器向客户端推送数据的技能,但它们之间存在一些关键区别。下面我将详细解释这两种技能的差别之处:
长链接(Long Polling)

长链接是一种实现服务器推送数据到客户端的技能,它基于HTTP请求/响应模型。在这种模式下,客户端发起一个HTTP请求,服务器在没有数据可发送的环境下会保持连接打开,直到有数据可发送大概超时。一旦服务器有数据要发送,它就会响应客户端的请求,并关闭连接。客户端接收到数据后立刻重新发起一个新的请求,从而保持与服务器的“长链接”。
特点:


  • 客户端主动发起请求:客户端需要不停地向服务器发起请求以获取数据。
  • 服务器被动响应:服务器只在客户端请求时才发送数据。
  • 连接短暂:虽然每个连接大概会持续一段时间,但每次请求竣事后连接会被关闭。
  • 实现简单:易于用现有HTTP技能实现。
  • 兼容性好:险些所有欣赏器都支持HTTP请求/响应模型。
Server-Sent Events (SSE)

Server-Sent Events 是一种更为现代的技能,用于实现服务器向客户端的单向数据推送。SSE基于HTTP协议,但使用了一个持久的HTTP连接来维持客户端和服务端之间的通信。服务器可以主动向客户端发送数据,而不需要等候客户端的请求。
特点


  • 服务器主动推送:服务器可以主动向客户端发送数据,而不需要客户端发起请求。
  • 持久连接:客户端和服务端之间建立了一个持久的连接,直到客户端或服务器关闭该连接。
  • 格式特定:SSE使用特定的格式来发送数据,包括data:字段和空行作为分隔符。
  • 资源服从高:由于连接是持久的,因此淘汰了建立连接的开销。
  • 实现复杂度适中:虽然比长链接稍微复杂,但现代欣赏器和服务器框架提供了精良的支持。
比力



  • 实时性:SSE提供更好的实时性,因为它不需要客户端不停发起请求。
  • 性能:SSE在性能上通常优于长链接,因为它避免了重复建立连接的开销。
  • 实现复杂度:SSE需要客户端和服务端双方的支持,而长链接可以更容易地在现有的HTTP基础设施上实现。
  • 兼容性:SSE在现代欣赏器中得到了广泛支持,但对于一些旧版欣赏器大概不适用;长链接则具有更好的向后兼容性。
总结

选择哪种技能取决于你的具体需求。如果你的应用需要较低延迟的数据推送,并且可以依靠现代欣赏器和服务器环境,那么SSE是一个不错的选择。如果你需要更广泛的欣赏器兼容性,并且对实时性要求不是特殊高,那么长链接大概更得当你。
2.通过okhttp调用SSE流式接口并流式返回给前端

环境要求



  • Spring Framework 5.0
  • Jdk1.8
使用okhttp相关依靠

  1. <dependency>
  2.     <groupId>com.squareup.okhttp3</groupId>
  3.     <artifactId>okhttp</artifactId>
  4.     <version>4.2.0</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>com.squareup.okhttp3</groupId>
  8.     <artifactId>okhttp-sse</artifactId>
  9.     <version>4.2.0</version>
  10. </dependency>
复制代码
示例

  1. @GetMapping(value = "/test1", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
  2.     public SseEmitter SseTest1() {
  3.         SseEmitter sseEmitter = new SseEmitter();
  4.         String prompt = "";
  5.         String url = "";
  6.         FormBody formBody = new FormBody.Builder().add("prompt", prompt).build();
  7.         Request request = new Request.Builder().url(url).post(formBody).build();
  8.         // 使用EventSourceListener处理来自服务器的SSE事件
  9.         EventSourceListener listener = new EventSourceListener() {
  10.             @Override
  11.             public void onOpen(@NotNull EventSource eventSource, @NotNull Response response) {
  12.                 log.info("Connection opened.");
  13.             }
  14.             @Override
  15.             public void onClosed(@NotNull EventSource eventSource) {
  16.                 log.info("Connection closed.");
  17.                 sseEmitter.complete();
  18.             }
  19.             @Override
  20.             public void onEvent(@NotNull EventSource eventSource, @Nullable String id, @Nullable String type, @NotNull String data) {
  21.                 try {
  22.                     JSONObject jsonObject = JSONUtil.parseObj(data);
  23.                     String event = jsonObject.getStr("event");
  24.                     if ("message".equals(event)) {
  25.                         sseEmitter.send(jsonObject.getStr("answer"));
  26.                     }
  27.                 } catch (Exception e) {
  28.                     log.error("推送数据失败", e);
  29.                 }
  30.             }
  31.             @Override
  32.             public void onFailure(@NotNull EventSource eventSource, @Nullable Throwable t, @Nullable Response response) {
  33.                 log.error("Connection failed.", t);
  34.                 sseEmitter.completeWithError(t);
  35.             }
  36.         };
  37.         OkHttpClient client = new OkHttpClient.Builder().connectTimeout(10, TimeUnit.SECONDS).writeTimeout(50, TimeUnit.SECONDS).readTimeout(10, TimeUnit.MINUTES).build();
  38.         EventSource.Factory factory = EventSources.createFactory(client);
  39.         factory.newEventSource(request, listener);
  40.         return sseEmitter;
  41.     }
复制代码
注意


  • 该接口需为_Get_请求,ContentType为 text/event-stream
  • SseEmitter 是Spring Framework 5.0引入的一个新特性,用于简化Server-Sent Events (SSE) 的实现。它提供了一种简单的方式来发送事件数据到客户端,特殊适用于构建实时数据推送的应用步调。
3. 如果Spring Framework 低于5.0,可使用Servlet 3.0举行流式返回

使用_AsyncContext_:Servlet 3.0 引入了异步支持,答应Servlet在差别的线程中处置惩罚请求。你可以使用AsyncContext来启动一个异步线程,在该线程中发送SSE事件。
配置_async-supported_
使用_AsyncContext_前需配置_async-supported_
async-supported元素用于指定Servlet是否支持异步处置惩罚。这个配置通常是在摆设描述符 web.xml 文件中举行设置的。
配置示例
  1. <web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"
  2.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
  4.          version="3.1">
  5.     <servlet>
  6.         <servlet-name>MyServlet</servlet-name>
  7.         <servlet-class>com.example.MyServlet</servlet-class>
  8.         <async-supported>true</async-supported>
  9.     </servlet>
  10.     <servlet-mapping>
  11.         <servlet-name>MyServlet</servlet-name>
  12.         <url-pattern>/myServlet</url-pattern>
  13.     </servlet-mapping>
  14. </web-app>
复制代码
后端代码示例
  1.         @GetMapping("/test3")
  2.     public void sseTest3(HttpServletRequest req, HttpServletResponse resp) {
  3.             // text/event-stream 是一个特殊的MIME类型,用于定义Server-Sent Events (SSE)。它告诉浏览器这个响应是SSE流,浏览器应该以这种方式处理接收到的数据。
  4.         resp.setContentType("text/event-stream");
  5.         resp.setCharacterEncoding("UTF-8");
  6.         // 这行代码设置了Cache-Control HTTP头部字段,值为no-cache。这意味着浏览器不应该缓存此响应。对于SSE来说,这是很重要的,因为我们希望实时更新数据,而不希望浏览器缓存旧的数据。
  7.         resp.setHeader("Cache-Control", "no-cache");
  8.         // 这行代码设置了Connection HTTP头部字段,值为keep-alive。这意味着客户端和服务器之间的TCP连接在响应完成后保持打开状态,以便后续的SSE事件可以通过同一个连接发送。这对于持续的数据流非常重要,因为它减少了建立新连接的开销。
  9.         resp.setHeader("Connection", "keep-alive");
  10.         try {
  11.             // 创建一个AsyncContext对象并开启异步处理流程
  12.             AsyncContext asyncContext = req.startAsync(req, resp);
  13.             asyncContext.setTimeout(10 * 60 * 1000);
  14.             PrintWriter writer = asyncContext.getResponse().getWriter();
  15.             String prompt = "";
  16.             String url = "";
  17.             FormBody formBody = new FormBody.Builder().add("prompt", prompt).build();
  18.             Request request = new Request.Builder().url(url).post(formBody).build();
  19.             // 使用EventSourceListener处理来自服务器的SSE事件
  20.             EventSourceListener listener = new EventSourceListener() {
  21.                 @Override
  22.                 public void onOpen(@NotNull EventSource eventSource, @NotNull Response response) {
  23.                     log.info("Connection opened.");
  24.                 }
  25.                 @Override
  26.                 public void onClosed(@NotNull EventSource eventSource) {
  27.                     log.info("Connection closed.");
  28.                     writer.write("data: __stop__
  29. ");
  30.                     writer.flush();
  31.                     asyncContext.complete();
  32.                 }
  33.                 @Override
  34.                 public void onEvent(@NotNull EventSource eventSource, @Nullable String id, @Nullable String type, @NotNull String data) {
  35.                     try {
  36.                         JSONObject jsonObject = JSONUtil.parseObj(data);
  37.                         String event = jsonObject.getStr("event");
  38.                         if ("message".equals(event)) {
  39.                             String answer = jsonObject.getStr("answer");
  40.                             log.info("message: {}", answer);
  41.                             writer.write("data: " + answer + "
  42. ");
  43.                             writer.flush();
  44.                         }
  45.                     } catch (Exception e) {
  46.                         log.error("推送数据失败", e);
  47.                     }
  48.                 }
  49.                 @Override
  50.                 public void onFailure(@NotNull EventSource eventSource, @Nullable Throwable t, @Nullable Response response) {
  51.                     log.error("Connection failed.", t);
  52.                     asyncContext.complete();
  53.                 }
  54.             };
  55.             OkHttpClient client = new OkHttpClient.Builder().connectTimeout(10, TimeUnit.SECONDS).writeTimeout(50, TimeUnit.SECONDS).readTimeout(10, TimeUnit.MINUTES).build();
  56.             EventSource.Factory factory = EventSources.createFactory(client);
  57.             factory.newEventSource(request, listener);
  58.         } catch (IOException e) {
  59.             e.printStackTrace();
  60.         }
  61.     }
复制代码
注意
返回数据格式:
  1. // 以data: 开头  /n/n结束
  2. "data: xxxxx /n/n"
复制代码
4. 前端调用SSE接口

方式1 使用JavaScript的 EventSource API

  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>SSE Example</title>
  6. </head>
  7. <body>
  8.     <div id="events"></div>
  9.     <script>
  10.         const source = new EventSource('/sse');
  11.         source.onmessage = function(event) {
  12.             const data = JSON.parse(event.data);
  13.             // 约定一个结束标识
  14.             if(data == '__stop__') {
  15.                     source.close()
  16.                     return
  17.             }
  18.             document.getElementById('events').innerHTML += `<p>${data.message}</p>`;
  19.         };
  20.         source.onerror = function(error) {
  21.             console.error('Error occurred:', error);
  22.             source.close();
  23.         };
  24.     </script>
  25. </body>
  26. </html>
复制代码
注意

  • 后端返回需返回完备消息的对象(包括换行符),例:{“data”: “哈哈哈/n/n”},如果后端将data取出,则会导致换行符丢失!
  • EventSource 只支持Get请求,如果请求参数过长会导致调用失败!
方式2 使用 fetchEventSource 插件

安装插件
  1. npm install --save @microsoft/fetch-event-source
复制代码
简单示例
  1. // 导入依赖
  2. import { fetchEventSource } from '@microsoft/fetch-event-source';
  3. send() {
  4.         const params = {
  5.                 "prompt" : ""
  6.         }
  7.         const vm = this;
  8.         const ctrlAbout = new window.AbortController();
  9.         const { signal } = ctrlAbout;
  10.         fetchEventSource(Url, {
  11.           method: 'POST',
  12.           headers: {
  13.             "Content-Type": 'application/json',
  14.             "Accept": 'text/event-stream',
  15.             "X-Requested-With": 'XMLHttpRequest'
  16.           },
  17.           body: JSON.stringify(params),
  18.           signal: ctrl.signal, // AbortSignal
  19.           openWhenHidden: true, // 取消visibilityChange事件
  20.           onmessage(event) {
  21.              console.info(event.data);
  22.              // 在这里操作流式数据
  23.              const message = JSON.parse(event.data)
  24.              vm.content += message.data
  25.              // 保证在打字输出时滚动条在最下方
  26.              vm.$nextTick(() => {
  27.                      const contentEl = vm.$refs.content.$el
  28.                      contentEl.scrollTop = contentEl.scrollHeight - contentEL.clientHeight
  29.              })
  30.           },
  31.           onclose(e) {
  32.              // 关闭流
  33.              // 中断流式返回
  34.              ctrl.abort()
  35.           }
  36.           onerror(error) {
  37.             // 返回流报错
  38.                 console.info(error);
  39.                 // 中断流式返回
  40.                 ctrl.abort()
  41.                 throw err // 直接抛出错误,避免反复调用
  42.           }
  43.         })
  44. }
复制代码
注意

  • 传参时需注意参数类型为json字符串
5. 使用原生的http调用SSE流式接口

示例
  1. @GetMapping("/test2")
  2.     public void SseTest2() {
  3.         String urlAddr = "";
  4.         BufferedReader reader = null;
  5.         try {
  6.             URL url = new URL(urlAddr);
  7.             // 建立链接
  8.             HttpURLConnection connection = (HttpURLConnection) url.openConnection();
  9.             connection.setRequestMethod("POST");
  10.             connection.setRequestProperty("Accept", "text/event-stream");
  11.             connection.setRequestProperty("Content-type", "application/json; charset=UTF-8");
  12.             connection.setRequestProperty("Cache-Control", "no-cache");
  13.             connection.setRequestProperty("Connection", "keep-alive");
  14.             // 允许输入和输出
  15.             connection.setDoInput(true);
  16.             connection.setDoOutput(true);
  17.             // 设置超时为0,表示无限制
  18.             connection.setConnectTimeout(0);
  19.             connection.setReadTimeout(0);
  20.             // 传参
  21.             String params = "prompt=哈哈哈哈";
  22.             // 写入POST数据
  23.             DataOutputStream out = new DataOutputStream(connection.getOutputStream());
  24.             out.write(params.getBytes(StandardCharsets.UTF_8));
  25.             out.flush();
  26.             out.close();
  27.             // 读取SSE事件
  28.             reader = new BufferedReader(new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8));
  29.             StringBuilder eventBuilder = new StringBuilder();
  30.             String line;
  31.             while ((line = reader.readLine()) != null) {
  32.                 System.out.println(line);
  33.             }
  34.             reader.close();
  35.             // 断开链接
  36.             connection.disconnect();
  37.         } catch (Exception e) {
  38.             e.printStackTrace();
  39.         } finally {
  40.             IoUtil.close(reader);
  41.         }
  42.     }
复制代码
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

正序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

九天猎人

金牌会员
这个人很懒什么都没写!

标签云

快速回复 返回顶部 返回列表