文章目录
- 从传统 HTTP 到 SSE：实时通信的演进之路
- 传统 HTTP 的局限性
- Server-Sent Events的诞生：服务器主动推送
- SSE 工作原理
- SSE 的局限性
- SSE 的适用场景
- SSE VS WebSocket
- 实时Web核心诉求
- 实战项目中的SSE
-
从传统 HTTP 到 SSE：实时通信的演进之路
- 在Web 应用中，实时数据推送已成为许多场景的核心需求，例如实时通知、股票行情、在线聊天和物联网设备监控。然而，传统的 HTTP 请求-响应模式在实时性上存在天然缺陷。 Server-Sent Events（SSE） 实现高效、轻量的服务器主动推送。
传统 HTTP 的局限性
- 请求-响应模式：传统的 HTTP 协议基于客户端主动请求、服务器被动响应的模式。
setInterval(() => {fetch('/api/data').then(response => response.json()).then(data => updateUI(data));
}, 5000);
- 高延迟：数据更新依赖客户端轮询频率（如 5 秒一次），无法实时获取最新状态。
- 资源浪费：频繁的请求消耗服务器和网络资源，尤其当数据未变化时。
- 长轮询（Long Polling）：为了优化轮询效率，长轮询允许服务器在数据就绪前保持连接打开。
function longPoll() {fetch('/api/long-poll').then(response => response.json()).then(data => {updateUI(data);longPoll(); });
}
- 尽管减少了无效请求，但长轮询仍需要反复建立连接，且复杂度较高。
Server-Sent Events的诞生：服务器主动推送
- SSE（Server-Sent Events） 是一种基于 HTTP 的轻量协议，支持服务器向客户端单向推送事件流。其核心思想是：
- 长连接：客户端与服务器建立一次 HTTP 连接后，保持打开状态。
- 事件驱动：服务器可在任意时刻推送数据，客户端通过监听事件实时接收。
SSE 工作原理
- 客户端：通过 EventSource API 订阅服务器事件流。
const eventSource = new EventSource('/sse');eventSource.onmessage = (event) => {console.log('收到数据:', event.data);
};
- 服务器：以 text/event-stream 格式持续发送事件。
HTTP/1.1 200 OK
Content-Type: text/event-streamdata: 这是第一条消息\n\n
id: 1\n
event: status\n\n
SSE 的局限性
- 单向通信：不支持客户端向服务器发送数据（需配合其他 HTTP 请求）。
- 浏览器兼容性：IE 及旧版 Edge 不支持，但可通过 Polyfill 解决。
- 最大连接数：浏览器对同一源的 SSE 连接数有限制（通常 6 个）。
SSE 的适用场景
| 场景 | 传统HTTP | SSE 方案 |
|---|
| 实时通知 | 高延迟，资源浪费 | 即时推送，低延迟 |
| 股票行情 | 轮询导致数据滞后 | 每秒推送多次价格更新 |
| 日志监控 | 需手动刷新页面 | 实时滚动显示日志 |
| 新闻头条 | 用户错过更新 | 新文章自动推送到客户端 |
SSE VS WebSocket
- SSE与WebSocket的相同点：都是用来建立浏览器与服务器之间的通信渠道。
- SSE与WebSocket的相同点的区别：
| WebSocket | SSE |
|---|
| 通道类型 | 双向全双工 | 单向通道（服务器->浏览器） |
| 协议类型 | 独立协议（ws://）协议 | HTTP协议 |
| 复杂度 | 需处理握手、帧协议 | 默认支持 |
| 断线重连 | 需手动实现 | 自动处理 |
| 消息自定义类型 | 不支持 | 支持 |
| 适用场景 | 服务器主导的推送场景 | 双向交互（如在线游戏） |
实时Web核心诉求
- 实时 Web 技术的核心诉求：更低延迟、更高效率、更简单的实现。
- SSE 凭借其轻量级、基于 HTTP 的特点，成为服务器推送场景的首选方案。在 Spring Boot 中，通过 S s e E m i t t e r SseEmitter SseEmitter 可以快速构建实时功能，而无需引入复杂的第三方库。
实战项目中的SSE
前端
methods: {initSSE(userName) {if (!window.EventSource) {console.log("浏览器不支持SSE")return}const source = new EventSource(`http://localhost:8443/sse/connect?userId=${userName}`)console.log("连接用户=", userName)this.currentUserName = userNamesource.addEventListener('open', () => {console.log("建立连接...")})source.addEventListener('add', (e) => {console.log("add事件...", e.data)const receiveMsg = e.dataif (!this.botMsgId) {this.botMsgId = this.generateRandomId(12)const newBotContent = {id: "temp",content: receiveMsg,userName: '家庭医生',chatType: 'bot',botMsgId: this.botMsgId}this.chatList.push(newBotContent)} else {const chatItem = this.chatList.find(item => item.botMsgId === this.botMsgId)if (chatItem) {chatItem.content += receiveMsg}}this.$nextTick(() => {this.scrollToBottom()})})source.addEventListener('finish', () => {console.log("finish事件...")this.botMsgId = nullthis.scrollToBottom()})source.addEventListener('error', (e) => {console.log("error事件...", e)if (e.readyState === EventSource.CLOSED) {console.log('connection is closed')}source.close()})},async getChatRecords(userName) {try {const result = await doctorApi.getRecords(userName)this.chatList = resultthis.$nextTick(() => {this.scrollToBottom()})} catch (err) {console.error('获取聊天记录失败:', err)}},generateRandomId(length) {const characters = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789'let result = ''for (let i = 0; i < length; i++) {result += characters.charAt(Math.floor(Math.random() * characters.length))}return result}
import request from '@/api/request'export const doctorApi = {getRecords(userName) {return request({url: `/record/getRecordList?userName=${userName}`,method: 'get'})},doChat(chatData) {return request({url: '/ai/chat',method: 'post',data: chatData})}
}
后端
连接接口
@Slf4j
@RestController
@RequestMapping("/sse")
public class SSEController {@GetMapping(path = "/connect", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})public SseEmitter connect(@RequestParam String userId) {return SSEServer.connect(userId);}
}
public static SseEmitter connect(String userId) {SseEmitter sseEmitter = new SseEmitter(0L);sseEmitter.onCompletion(completionCallback(userId));sseEmitter.onError(errorCallback(userId));sseEmitter.onTimeout(timeoutCallback(userId));sseClients.put(userId, sseEmitter);log.info("当前创建新的SSE连接，用户ID为: {}", userId);onlineCounts.getAndIncrement();return sseEmitter;
}
获取聊天记录
@RestController
@RequestMapping("/record")
public class RecordController {@Resourceprivate ChatRecordService chatRecordService;@GetMapping("/getRecordList")public List<ChatRecord> getRecordList(@RequestParam String userName) {return chatRecordService.getChatRecordList(userName);}}
@Overridepublic List<ChatRecord> getChatRecordList(String userName) {QueryWrapper<ChatRecord> queryWrapper = new QueryWrapper<>();queryWrapper.eq("family_member", userName);queryWrapper.orderByAsc("chat_time");return chatRecordMapper.selectList(queryWrapper);}
聊天接口
@lombok.extern.slf4j.Slf4j
@Slf4j
@RestController()
@RequestMapping("/ai")
public class ChatController {@Resourceprivate AIService aiService;@PostMapping("/chat")public String chatWithDoctor(@RequestBody ChatEntity chatEntity) {log.info(chatEntity.toString());String currentUserName = chatEntity.getCurrentUserName();String message = chatEntity.getMessage();return aiService.chatWithDoctor(currentUserName,message);}
}
@Override
public String chatWithDoctor(String userName, String message) {if (message == null || message.isEmpty()) {return "message is empty";}chatRecordService.saveChatRecord(userName, message, ChatTypeEnum.USER);Prompt prompt = new Prompt(new UserMessage(message));log.info(prompt.toString());List<String> list = this.chatModel.stream(prompt).toStream().map(chatResponse -> {String text = chatResponse.getResult().getOutput().getText();SSEServer.sendMessage(userName, text, SSEMsgType.ADD);return text;}).toList();SSEServer.sendMessage(userName, "finish", SSEMsgType.FINISH);StringBuilder htmlRes= new StringBuilder();for (String s : list) {htmlRes.append(s);}chatRecordService.saveChatRecord(userName, htmlRes.toString(), ChatTypeEnum.BOT);return "success";
}
