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

weixin_44680858 2024-08-20 17:03:02 阅读 77

目录

1.SSE概述1.1 什么是是SSE2.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使用特定的格式来发送数据,包括<code>data:字段和空行作为分隔符。资源效率高:由于连接是持久的,因此减少了建立连接的开销。实现复杂度适中:虽然比长链接稍微复杂,但现代浏览器和服务器框架提供了良好的支持。

比较

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

总结

选择哪种技术取决于你的具体需求。如果你的应用需要较低延迟的数据推送,并且可以依赖现代浏览器和服务器环境,那么SSE是一个不错的选择。如果你需要更广泛的浏览器兼容性,并且对实时性要求不是特别高,那么长链接可能更适合你。

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

环境要求

Spring Framework 5.0Jdk1.8

使用okhttp相关依赖

<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处理来自服务器的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("推送数据失败", 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-streamSseEmitter 是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 文件中进行设置的。

配置示例

<web-app xmlns="http://xmlns.jcp.org/xml/ns/javaee"code>

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"code>

xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"code>

version="3.1">code>

<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处理来自服务器的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("推送数据失败", 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 {

}

}

注意

返回数据格式:

// 以data: 开头 /n/n结束

"data: xxxxx /n/n"

4. 前端调用SSE接口

方式1 使用JavaScript的 EventSource API

<!DOCTYPE html>

<html lang="en">code>

<head>

<meta charset="UTF-8">code>

<title>SSE Example</title>

</head>

<body>

<div id="events"></div>code>

<script>

const source = new EventSource('/sse');

source.onmessage = function(event) { -- -->

const data = JSON.parse(event.data);

// 约定一个结束标识

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取出,则会导致换行符丢失!EventSource 只支持Get请求,如果请求参数过长会导致调用失败!

方式2 使用 fetchEventSource 插件

安装插件

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

简单示例

// 导入依赖

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);

// 在这里操作流式数据

const message = JSON.parse(event.data)

vm.content += message.data

},

onclose(e) {

// 关闭流

// 中断流式返回

ctrl.abort()

}

onerror(error) {

// 返回流报错

console.info(error);

// 中断流式返回

ctrl.abort()

throw err // 直接抛出错误,避免反复调用

}

})

}

注意

传参时需注意参数类型为json字符串

5. 使用原生的http调用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);

// 设置超时为0,表示无限制

connection.setConnectTimeout(0);

connection.setReadTimeout(0);

// 传参

String params = "prompt=哈哈哈哈";

// 写入POST数据

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);

}

}



声明

本文内容仅代表作者观点,或转载于其他网站,本站不以此文作为商业用途
如有涉及侵权,请联系本站进行删除
转载本站原创文章,请注明来源及作者。