在现代 Web 应用中,实时数据传输已经成为一个不可忽视的需求,尤其是在聊天系统、实时数据展示、进度推送等场景中。
Spring 提供了 SseEmitter
来支持服务器端推送事件(Server-Sent Events, SSE),它通过 HTTP 协议允许服务器主动向客户端推送数据。与 WebSocket 不同,SSE 是单向通信,但它也提供了一种轻量、高效的实时数据流解决方案。
本文将从原理、实际应用、客户端和服务端代码示例、扩展应用场景等方面详细介绍 Spring SseEmitter
。
1. 原理解析
1.1 Server-Sent Events (SSE)
SSE 是一种基于 HTTP 协议的单向数据流技术,允许服务器通过持久化的 HTTP 连接将事件推送到客户端。其与 WebSocket 的主要区别是,SSE 是单向通信(从服务器到客户端),而 WebSocket 是全双工通信(服务器和客户端都可以发送数据)。
在 SSE 中,客户端通过 EventSource
对象与服务器建立连接,服务器将数据以流的方式推送到客户端。SSE 使用标准的 HTTP 协议,并且是基于文本的流,数据通常以 text/event-stream
类型的格式传输。
1.2 SseEmitter
SseEmitter
是 Spring 提供的一个类,用于实现 SSE 功能。它支持向客户端推送数据,并且能够处理长连接。SseEmitter
的主要功能包括:
- 异步发送数据:SSE 是长连接,
SseEmitter
可以在服务器端异步发送数据。 - 支持超时管理:
SseEmitter
可以设置超时,若客户端在指定时间内没有响应,连接会自动关闭。 - 异常处理:如果发生错误,可以通过
SseEmitter
完成错误通知和连接关闭。
2. 服务端代码实现
2.1 创建 SSE 连接
使用 Spring 的 SseEmitter
来向客户端推送实时事件非常简单。
我们通过 @GetMapping
注解来映射一个请求路径,返回一个 SseEmitter
实例。
@RestController
public class SseController {
@GetMapping(value = "/sync/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter syncStream() {
SseEmitter emitter = new SseEmitter();
// 模拟异步任务,发送实时数据
new Thread(() -> {
try {
for (int i = 0; i <= 100; i++) {
emitter.send("data: " + i + "% complete\n\n");
Thread.sleep(1000); // 模拟任务处理
}
emitter.complete(); // 任务完成,结束 SSE 流
} catch (Exception e) {
emitter.completeWithError(e); // 出现异常时结束流
}
}).start();
return emitter;
}
}
在上面的代码中,syncStream
方法创建了一个 SseEmitter
实例,并使用一个线程模拟了一个处理任务的过程。
每秒钟将当前进度推送给客户端,直到进度达到 100%。当任务完成时,通过 emitter.complete()
关闭连接。
SseEmitter.send(data)
: 用于向客户端推送数据。SseEmitter.complete()
: 用于标记推送完成,关闭连接。SseEmitter.completeWithError(exception)
: 如果发生异常,可以通过此方法结束连接并发送错误信息。
2.2 超时设置
SSE 连接可能会因为网络问题或其他原因被中断,Spring 提供了超时设置,确保连接不被无限期阻塞。
@GetMapping(value = "/sync/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter syncStream() {
SseEmitter emitter = new SseEmitter(30000L); // 设置超时时间为30秒
// 发送数据的代码同上...
return emitter;
}
3. 客户端代码实现
在客户端,我们可以使用原生的 JavaScript EventSource
对象来接收 SSE 流数据。EventSource
会自动管理连接,包括重试和重新连接等操作。
let eventSource = new EventSource("/sync/stream");
eventSource.onmessage = function(event) {
console.log("Received message: ", event.data);
};
eventSource.onerror = function(error) {
console.error("EventSource failed: ", error);
};
new EventSource("/sync/stream")
: 创建一个连接到指定 URL 的EventSource
实例。onmessage
: 每当服务器发送一个事件时,会触发该事件。onerror
: 当发生错误时,会触发该事件。
4. 扩展应用场景
4.1 实时任务进度推送
可以利用 SseEmitter
来推送后台任务的实时进度。
例如,在执行一个需要时间的任务时,可以定期向客户端推送进度更新。
@GetMapping("/task/progress")
public SseEmitter streamTaskProgress() {
SseEmitter emitter = new SseEmitter();
new Thread(() -> {
try {
for (int i = 0; i <= 100; i++) {
emitter.send("data: " + i + "% complete\n\n");
Thread.sleep(1000); // 模拟任务处理
}
emitter.complete(); // 任务完成,结束 SSE 流
} catch (Exception e) {
emitter.completeWithError(e); // 出现异常时结束流
}
}).start();
return emitter;
}
4.2 实时聊天系统
在实时聊天应用中,可以使用 SSE 向所有连接的客户端推送消息。
@GetMapping("/chat/{roomId}")
public SseEmitter streamChatMessages(@PathVariable String roomId) {
SseEmitter emitter = new SseEmitter();
chatService.addListener(roomId, message -> {
try {
emitter.send("data: " + message + "\n\n");
} catch (IOException e) {
emitter.completeWithError(e);
}
});
return emitter;
}
在这个例子中,
chatService.addListener(roomId, message -> {...})
是监听指定聊天室消息的逻辑,所有聊天信息都会通过 SseEmitter.send()
推送到客户端。
4.3 动态数据显示
SSE 适用于动态显示数据更新的场景,比如股票价格、天气预报等实时数据。
@GetMapping("/live-stock/{symbol}")
public SseEmitter streamStockPrice(@PathVariable String symbol) {
SseEmitter emitter = new SseEmitter();
stockService.addPriceListener(symbol, price -> {
try {
emitter.send("data: " + price + "\n\n");
} catch (IOException e) {
emitter.completeWithError(e);
}
});
return emitter;
}
在这个例子中,
stockService.addPriceListener(symbol, price -> {...})
监听股票的实时价格变化,并通过 SseEmitter.send()
将最新价格推送到客户端。
5. 总结
Spring 的 SseEmitter
提供了一种简洁且高效的方式来实现服务器向客户端推送实时事件。通过 SSE,我们可以轻松地在后台执行长时间任务,并将实时数据推送给前端应用。
SSE 适用于许多场景,如任务进度推送、实时消息通知、动态数据展示等。相较于 WebSocket,SSE 实现简单,且不需要额外的协议支持,适合轻量级的实时应用。
以上为个人经验,希望能给大家一个参考,也希望大家多多支持QQ沐编程。