SpringBoot / Vue 对SSE的基本使用（简单上手）

'# SpringBoot / Vue 对SSE的基本使用（简单上手）

一、背景与问题

在现代Web开发中，实时通信需求日益增长。传统的HTTP请求-响应模型无法满足实时数据推送需求，而WebSocket虽然能够实现双向通信，但其建立连接的复杂性和跨域限制使得其在某些场景下并不适用。

Server-Sent Events（SSE）作为HTML5引入的服务器向客户端推送数据的机制，提供了轻量级的实时通信方案。它基于HTTP协议，利用长连接保持通信，同时支持事件流格式（EventStream），在实时通知、数据推送等场景中具有独特优势。

本篇文章将深入解析SSE的原理，通过SpringBoot和Vue的完整案例，展示如何在实际开发中使用SSE技术，同时分析其适用场景、性能优化方法和常见问题。

二、基本原理

1. 工作机制

SSE的核心是通过HTTP长连接实现服务器向客户端的单向数据推送。其关键特征包括：

基于HTTP协议：无需额外协议支持，兼容性好
事件流格式：使用text/event-stream MIME类型传输数据
自动重连机制：客户端自动尝试重新连接
消息格式：支持自定义数据字段和事件类型

通信流程如下：

客户端发送请求 → 服务器保持连接 → 客户端接收事件流数据

2. 通信协议

SSE通信数据包格式如下：

event: notify
id: 1
data: {"type": "message", "content": "Hello, SSE!"}
retry: 5000

关键字段说明：

event：事件类型（可选）
id：事件ID（用于断线重连）
data：事件数据（JSON格式）
retry：重连间隔（单位：毫秒）

三、环境准备

1. 技术栈

后端：SpringBoot 2.7 + Java 17
前端：Vue 3 + TypeScript
数据库：MySQL 8.0（可选）

2. 依赖配置

SpringBoot项目中添加依赖：

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

Vue项目中安装依赖：

npm install eventsource

四、核心实现

1. SpringBoot服务端实现

1.1 创建SSE接口

@RestController
public class SseController {

    @GetMapping("/sse")
    public SseEmitter sse() {
        // 设置超时时间（默认30秒）
        return new SseEmitter(30_000L);
    }

    @PostMapping("/send")
    public void send(@RequestParam String id, @RequestBody Map<String, Object> data) {
        // 通过id找到对应的SseEmitter发送数据
        sseEmitterMap.get(id).send(SseEmitter.event()
                .name("message")
                .data(data)
                .id(id)
                .retry(5000)
        );
    }
}

关键点：

使用SseEmitter类创建长连接
通过send()方法发送事件数据
支持name字段指定事件类型
retry字段设置重连间隔

1.2 管理连接

@Singleton
public class SseEmitterManager {
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

    public void add(String id, SseEmitter emitter) {
        emitters.put(id, emitter);
    }

    public void remove(String id) {
        emitters.remove(id);
    }

    public SseEmitter get(String id) {
        return emitters.get(id);
    }
}

2. Vue客户端实现

2.1 基础使用

<template>
  <div>
    <h2>SSE 接收消息</h2>
    <ul>
      <li v-for="(msg, index) in messages" :key="index">{{ msg }}</li>
    </ul>
  </div>
</template>

<script>
import { ref, onMounted } from 'vue'
import { EventSource } from 'eventsource'

export default {
  setup() {
    const messages = ref([])
    
    onMounted(() => {
      const eventSource = new EventSource('http://localhost:8080/sse')
      
      eventSource.onmessage = (event) => {
        messages.value.push(event.data)
      }
      
      eventSource.onerror = (event) => {
        console.error('SSE连接异常:', event)
      }
    })
    
    return { messages }
  }
}
</script>

关键点：

使用EventSource建立连接
onmessage处理普通消息
onerror处理连接异常
自动重连机制由浏览器实现

2.2 带事件类型处理

<script>
export default {
  setup() {
    const messages = ref([])
    
    onMounted(() => {
      const eventSource = new EventSource('http://localhost:8080/sse')
      
      eventSource.addEventListener('notify', (event) => {
        const data = JSON.parse(event.data)
        messages.value.push(`通知: ${data.content}`)
      })
      
      eventSource.onerror = (event) => {
        console.error('SSE连接异常:', event)
      }
    })
  }
}
</script>

3. 错误处理与性能优化

3.1 错误处理示例

@PostMapping("/send")
public void send(@RequestParam String id, @RequestBody Map<String, Object> data) {
    SseEmitter emitter = sseEmitterMap.get(id);
    if (emitter == null || emitter.isCompleted()) {
        throw new IllegalStateException("连接已断开");
    }
    emitter.send(SseEmitter.event()
            .name("message")
            .data(data)
            .id(id)
            .retry(5000)
    );
}

3.2 性能优化

使用连接池管理SseEmitter
设置合理的超时时间（默认30秒）
对异常连接进行清理
使用Redis存储连接信息（分布式场景）

五、完整案例：实时通知系统

1. 项目结构

src
├── main
│   ├── java
│   │   └── com.example.sse
│   │       ├── controller
│   │       │   └── SseController.java
│   │       └── service
│   │           └── SseService.java
│   └── resources
│       └── application.yml
└── test

2. 核心代码

2.1 控制器

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

    @Autowired
    private SseService sseService;

    @GetMapping
    public SseEmitter sse() {
        return sseService.createSseEmitter();
    }

    @PostMapping
    public void send(@RequestParam String id, @RequestBody Map<String, Object> data) {
        sseService.send(id, data);
    }
}

2.2 服务层

@Service
public class SseService {

    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

    public SseEmitter createSseEmitter() {
        SseEmitter emitter = new SseEmitter(30_000L);
        emitters.put(UUID.randomUUID().toString(), emitter);
        emitter.onCompletion(() -> emitters.remove(emitter.getId()));
        return emitter;
    }

    public void send(String id, Map<String, Object> data) {
        SseEmitter emitter = emitters.get(id);
        if (emitter != null && !emitter.isCompleted()) {
            emitter.send(SseEmitter.event()
                    .name("notification")
                    .data(data)
                    .id(id)
                    .retry(5000)
            );
        }
    }
}

2.3 前端页面

<template>
  <div>
    <h2>实时通知</h2>
    <div v-if="notifications.length">
      <h3>最新通知:</h3>
      <p>{{ notifications[notifications.length - 1] }}</p>
    </div>
    <button @click="sendNotification">发送通知</button>
  </div>
</template>

<script>
export default {
  data() {
    return {
      notifications: []
    }
  },
  mounted() {
    this.connectSSE();
  },
  methods: {
    connectSSE() {
      const eventSource = new EventSource('http://localhost:8080/sse');
      
      eventSource.addEventListener('notification', (event) => {
        this.notifications.push(event.data);
      });
      
      eventSource.onerror = (event) => {
        console.error('SSE连接异常:', event);
      };
    },
    sendNotification() {
      fetch('http://localhost:8080/sse/send', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          id: '123',
          content: '这是测试通知'
        })
      });
    }
  }
}
</script>

3. 运行效果

启动SpringBoot应用
打开Vue页面，连接SSE服务
点击"发送通知"按钮，服务器将推送消息到客户端
页面实时显示通知内容

六、源码解析

1. SpringBoot的SseEmitter机制

SseEmitter内部使用SseEventSource管理连接，关键逻辑如下：

public class SseEmitter {
    private final SseEventSource eventSource;
    private final int timeout;
    
    public SseEmitter(int timeout) {
        this.timeout = timeout;
        this.eventSource = new SseEventSource(timeout);
    }
    
    public void send(SseEvent event) {
        eventSource.send(event);
    }
}

2. 事件流数据格式

SSE数据包经过序列化后，格式为：

data: {"type":"notification","content":"测试消息"}
event: notification
id: 123
retry: 5000

3. 客户端EventSource实现

浏览器端的EventSource实现包含以下关键逻辑：

class EventSource {
    constructor(url) {
        this.url = url;
        this.xmlHttpRequest = new XMLHttpRequest();
        this.xmlHttpRequest.open('GET', this.url, true);
        this.xmlHttpRequest.setRequestHeader('Accept', 'text/event-stream');
        this.xmlHttpRequest.onreadystatechange = () => {
            if (this.xmlHttpRequest.readyState === 4) {
                this.handleResponse();
            }
        };
        this.xmlHttpRequest.onmessage = (event) => {
            this.onmessage(event);
        };
        this.xmlHttpRequest.onerror = (event) => {
            this.onerror(event);
        };
        this.xmlHttpRequest.send();
    }
}

七、进阶使用

1. 事件类型管理

@PostMapping
public void send(String id, Map<String, Object> data) {
    SseEmitter emitter = emitters.get(id);
    if (emitter != null && !emitter.isCompleted()) {
        emitter.send(SseEmitter.event()
                .name("user:status")
                .data(data)
                .id(id)
                .retry(5000)
        );
    }
}

2. 连接管理优化

使用Redis存储连接信息：

public void send(String id, Map<String, Object> data) {
    SseEmitter emitter = redisSseEmitter.get(id);
    if (emitter != null && !emitter.isCompleted()) {
        emitter.send(SseEmitter.event()
                .name("user:status")
                .data(data)
                .id(id)
                .retry(5000)
        );
    }
}

3. 安全增强

@PostMapping
public void send(@RequestParam String id, @RequestBody Map<String, Object> data) {
    if (!SecurityUtils.isAuthorized(id)) {
        throw new UnauthorizedException("无权限发送消息");
    }
    SseEmitter emitter = emitters.get(id);
    if (emitter != null && !emitter.isCompleted()) {
        emitter.send(SseEmitter.event()
                .name("user:status")
                .data(data)
                .id(id)
                .retry(5000)
        );
    }
}

八、性能与工程实践

1. 性能优化策略

优化项	方法	效果
连接复用	使用连接池	降低资源消耗
超时设置	设置合理超时时间	防止资源泄露
异常处理	定期检查连接状态	提高稳定性
负载均衡	使用Nginx反向代理	提升并发能力

2. 例外处理方案

public void send(String id, Map<String, Object> data) {
    SseEmitter emitter = emitters.get(id);
    if (emitter == null || emitter.isCompleted()) {
        // 记录日志并尝试重连
        log.warn("连接已断开: {}", id);
        return;
    }
    emitter.send(SseEmitter.event()
            .name("user:status")
            .data(data)
            .id(id)
            .retry(5000)
    );
}

3. 安全风险分析

CSRF攻击：需在请求中加入防伪令牌
身份验证：需要对接认证系统（如JWT）
数据泄露：需加密敏感数据
DDoS防护：需限制并发连接数

九、常见问题与踩坑

1. 常见错误

问题	原因	解决方案
连接断开	服务器端未正确关闭	使用`onCompletion`回调
未收到数据	未设置正确的Content-Type	在响应头中设置`text/event-stream`
重连失败	未正确处理异常	添加`onerror`回调
数据解析错误	数据格式不规范	严格校验JSON格式

2. 常见坑点

2.1 超时处理

错误示例：

new SseEmitter(30_000L) // 超时30秒

改进方案：

new SseEmitter(30_000L)
    .onCompletion(() -> {
        sseEmitterMap.remove(id);
    })

2.2 重复事件ID