当前位置：首页 > article >正文

springboot+vue+SseEmitter数据流推送实战

article 2024/11/18 4:28:12

业务场景

SseEmitter介绍

SseEmitter 是 Spring Boot 中用于实现服务器发送事件（Server-Sent Events, SSE）的一种机制。SSE 允许服务器向客户端推送实时数据，而不需要客户端频繁地发起请求。这对于实现实时通知、更新等场景非常有用。

SseEmitter与WebSocket区别

1. 单向 vs 双向通信

SSE (Server-Sent Events):
- 单向通信: SSE 只支持服务器向客户端推送数据，客户端不能通过同一个连接向服务器发送数据。
- 适用场景: 适用于只需要服务器向客户端推送数据的场景，如实时通知、股票行情、新闻更新等。
WebSocket:
- 双向通信: WebSocket 支持全双工通信，服务器和客户端都可以通过同一个连接发送数据。
- 适用场景: 适用于需要双向实时通信的场景，如在线聊天、多人协作编辑、实时游戏等。

2. 连接建立

SSE:
- 基于 HTTP: SSE 使用 HTTP 协议建立连接，客户端通过普通的 HTTP GET 请求连接到服务器。
- 简单性: 建立连接的过程相对简单，不需要额外的握手步骤。
WebSocket:
- 基于 WebSocket 协议: WebSocket 使用独立的协议，需要通过 WebSocket 握手建立连接。
- 复杂性: 建立连接的过程稍微复杂一些，需要客户端和服务器进行握手协商。

3. 数据格式

SSE:
- 文本数据: SSE 只能传输文本数据，数据格式通常是简单的文本或 JSON。
- 事件类型: 支持自定义事件类型，可以通过 event 字段指定。
WebSocket:
- 二进制/文本数据: WebSocket 支持传输二进制数据和文本数据，灵活性更高。
- 帧结构: 数据以帧的形式传输，支持多种子协议。

4. 连接保持

SSE:
- 长轮询: SSE 连接在数据发送完毕后会自动关闭，客户端需要重新建立连接。可以通过设置 retry 字段来控制重连时间。
- 自动重连: 浏览器会自动处理重连，但可能会有短暂的中断。
WebSocket:
- 持久连接: WebSocket 连接一旦建立，会一直保持打开状态，除非显式关闭或发生错误。
- 手动重连: 需要手动处理重连逻辑。

5. 浏览器支持

SSE:
- 广泛支持: SSE 在现代浏览器中得到了广泛支持，包括 Chrome、Firefox、Safari 和 Edge。
- 兼容性: 一些较旧的浏览器可能不支持 SSE，但可以通过 polyfill 来实现兼容。
WebSocket:
- 广泛支持: WebSocket 在现代浏览器中也得到了广泛支持，包括 Chrome、Firefox、Safari 和 Edge。
- 兼容性: 一些较旧的浏览器可能不支持 WebSocket，但可以通过 Flash 或其他技术来实现兼容。

6. 性能和资源消耗

SSE:
- 轻量级: SSE 的实现相对简单，资源消耗较低。
- 服务器负载: 由于连接会在数据发送完毕后关闭，服务器负载相对较小。
WebSocket:
- 高性能: WebSocket 由于支持持久连接，可以实现更低的延迟和更高的性能。
- 服务器负载: 由于连接一直保持打开状态，服务器需要管理更多的连接，可能会增加服务器负载。

总结

SSE 适用于简单的、单向的实时数据推送场景，实现简单，资源消耗低。
WebSocket 适用于复杂的、双向的实时通信场景，支持二进制数据，性能高，但实现复杂，资源消耗较高。
选择哪种技术取决于你的具体需求。如果你只需要服务器向客户端推送数据，且数据量不大，SSE 是一个很好的选择。如果你需要双向通信或传输大量数据，WebSocket 更适合。

入门教程

https://www.ruanyifeng.com/blog/2017/05/server-sent_events.html

实战案例

添加依赖

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

编写Controller

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@RestController
@RequestMapping("/sse")
public class SseController {

    private final ExecutorService executorService = Executors.newFixedThreadPool(10);

    @GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter connect() {
        SseEmitter sseEmitter = new SseEmitter(Long.MAX_VALUE); // 设置超时时间为 Long.MAX_VALUE

        // 异步处理消息发送
        executorService.submit(() -> {
            try {
                for (int i = 1; i <= 10; i++) {
                    sseEmitter.send(SseEmitter.event()
                            .id(String.valueOf(i))
                            .name("greeting")
                            .data("Hello, message " + i));
                    Thread.sleep(1000); // 模拟延迟
                }
                sseEmitter.complete(); // 完成发送
            } catch (IOException | InterruptedException e) {
                sseEmitter.completeWithError(e); // 发生错误时完成
            }
        });

        return sseEmitter;
    }
}

编写前端页面

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SSE Example</title>
</head>
<body>
    <h1>Server-Sent Events Example</h1>
    <ul id="messages"></ul>

    <script>
        const eventSource = new EventSource('/sse/connect');

        eventSource.onmessage = function(event) {
            const messages = document.getElementById('messages');
            const li = document.createElement('li');
            li.textContent = 'Message: ' + event.data;
            messages.appendChild(li);
        };

        eventSource.onerror = function(error) {
            console.error('EventSource failed:', error);
            eventSource.close();
        };
    </script>
</body>
</html>