Java调用SSE流式接口，并流式返回给前端实现打字输出效果 - ToB企服应用市场:ToB评测及商务社交产业平台

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

复制代码

<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) {
// text/event-stream 是一个特殊的MIME类型，用于定义Server-Sent Events (SSE)。它告诉浏览器这个响应是SSE流，浏览器应该以这种方式处理接收到的数据。
resp.setContentType("text/event-stream");
resp.setCharacterEncoding("UTF-8");
// 这行代码设置了Cache-Control HTTP头部字段，值为no-cache。这意味着浏览器不应该缓存此响应。对于SSE来说，这是很重要的，因为我们希望实时更新数据，而不希望浏览器缓存旧的数据。
resp.setHeader("Cache-Control", "no-cache");
// 这行代码设置了Connection HTTP头部字段，值为keep-alive。这意味着客户端和服务器之间的TCP连接在响应完成后保持打开状态，以便后续的SSE事件可以通过同一个连接发送。这对于持续的数据流非常重要，因为它减少了建立新连接的开销。
resp.setHeader("Connection", "keep-alive");
try {
// 创建一个AsyncContext对象并开启异步处理流程
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__
");
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 + "
");
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();
}
}

复制代码

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

复制代码

<!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);
// 约定一个结束标识
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>

复制代码

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

复制代码

// 导入依赖
import { fetchEventSource } from '@microsoft/fetch-event-source';
send() {
const params = {
"prompt" : ""
}
const vm = this;
const ctrlAbout = new window.AbortController();
const { signal } = ctrlAbout;
fetchEventSource(Url, {
method: 'POST',
headers: {
"Content-Type": 'application/json',
"Accept": 'text/event-stream',
"X-Requested-With": 'XMLHttpRequest'
},
body: JSON.stringify(params),
signal: ctrl.signal, // AbortSignal
openWhenHidden: true, // 取消visibilityChange事件
onmessage(event) {
console.info(event.data);
// 在这里操作流式数据
const message = JSON.parse(event.data)
vm.content += message.data
// 保证在打字输出时滚动条在最下方
vm.$nextTick(() => {
const contentEl = vm.$refs.content.$el
contentEl.scrollTop = contentEl.scrollHeight - contentEL.clientHeight
})
},
onclose(e) {
// 关闭流
// 中断流式返回
ctrl.abort()
}
onerror(error) {
// 返回流报错
console.info(error);
// 中断流式返回
ctrl.abort()
throw err // 直接抛出错误，避免反复调用
}
})
}

复制代码

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

复制代码