本文主要是介绍Web服务端通过SSE推送消息给浏览器客户端的实现方案(附详细代码和仓库地址),希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
目录
- 1、SSE(Server-Sent Events)简介
- 2、SSE 的工作原理
- 3、SSE 与客户端轮询的区别和优势比较
- 区别
- 优势
- 4、SSE简单实现(单机应用Demo)
- 演示效果
- SSE-Demo仓库地址
- 下面直接贴代码:
- 前端实现:
- 后端实现:
- 5、SSE简单实现(分布式应用Demo)
- SSE-Demo仓库地址
- 关键代码
- 方案说明
1、SSE(Server-Sent Events)简介
Server-Sent Events (SSE) 是一种基于 HTTP 协议的服务器推送技术,允许服务器通过单个持久连接向客户端发送实时更新。客户端使用标准的 EventSource API 来接收服务器推送的事件,这种通信方式非常适合实时应用,如消息通知、股票行情更新、社交媒体更新等。
2、SSE 的工作原理
- 单向连接:SSE 建立的是单向通道,即服务器向客户端推送数据,客户端只能接收,不能发送。
- 持久连接:SSE 使用的是长连接(Long Polling),即连接一旦建立,将会持续存在,直到客户端或服务器关闭连接。
- 文本数据:SSE 通过 text/event-stream MIME 类型传输数据,数据是纯文本格式。
- 自动重连:如果连接中断,EventSource 会自动尝试重新连接,确保客户端能够接收后续的推送。
3、SSE 与客户端轮询的区别和优势比较
客户端轮询(Client Polling) 是一种传统的客户端从服务器请求数据的方式。客户端会定期向服务器发送请求,检查是否有新数据可用。
区别
连接方式:
SSE:建立后服务器主动推送数据,连接是持久的,数据在有更新时实时传递。
轮询:客户端定期发送请求获取数据,连接是间歇性的。
实时性:
SSE:数据几乎是实时推送的,延迟极低。
轮询:数据获取延迟取决于轮询的频率,频率高则延迟低,但频率低可能导致数据延迟。
网络和服务器负载:
SSE:由于是单个持久连接,减少了频繁的请求与响应开销,降低了服务器负载。
轮询:频繁的请求会增加服务器和网络的负担,尤其是在轮询频率较高时。
连接控制:
SSE:自动处理连接中断和重连,客户端实现简单。
轮询:需要客户端定期发起请求,且如果请求频率不当,可能导致资源浪费。
数据传输效率:
SSE:只在有数据更新时推送,传输效率高。
轮询:即使没有数据更新,客户端也会定期请求,效率低下。
优势
SSE 的优势:
更高效的网络和服务器资源利用率。
实时性更高,延迟更低。
实现简单,特别是在浏览器环境中,支持自动重连和事件处理。
适合需要频繁更新但客户端无需响应的场景。
客户端轮询的优势:
在不支持 SSE 的环境下仍然可以使用。
实现和理解相对简单,兼容性更好。
4、SSE简单实现(单机应用Demo)
演示效果
SSE-Demo仓库地址
https://github.com/deepjava-gm/SSE-Demo.git
下面直接贴代码:
前端实现:
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>SSE 用户消息推送 Demo</title><style>body {font-family: Arial, sans-serif;margin: 20px;}h1 {text-align: center;}.container {max-width: 600px;margin: 0 auto;padding: 20px;border: 1px solid #ccc;border-radius: 8px;}.form-group {margin-bottom: 15px;}label {display: block;margin-bottom: 5px;font-weight: bold;}input[type="text"] {width: 100%;padding: 8px;box-sizing: border-box;border: 1px solid #ccc;border-radius: 4px;}button {padding: 10px 20px;color: white;background-color: #007bff;border: none;border-radius: 4px;cursor: pointer;}button:hover {background-color: #0056b3;}#messages {margin-top: 20px;}.message {background-color: #f1f1f1;padding: 10px;margin-bottom: 10px;border-radius: 4px;}.status {margin-top: 10px;font-weight: bold;}.success {color: green;}.error {color: red;}</style>
</head>
<body><h1>SSE 用户消息推送 Demo</h1><div class="container"><!-- 连接部分 --><div class="form-group"><label for="userId">用户 ID:</label><input type="text" id="userId" placeholder="请输入您的用户 ID"><br><br><button onclick="connect()">连接</button></div><div id="connectionStatus" class="status"></div><!-- 消息发送部分 --><div class="form-group"><label for="targetUserId">目标用户 ID:</label><input type="text" id="targetUserId" placeholder="请输入目标用户 ID"></div><div class="form-group"><label for="message">消息内容:</label><input type="text" id="message" placeholder="请输入要发送的消息"></div><button onclick="sendMessage()">推送消息</button><!-- 消息显示部分 --><div id="messages"></div></div><script>let eventSource;let currentUserId = '';function connect() {const userId = document.getElementById('userId').value;const connectionStatus = document.getElementById('connectionStatus');if (!userId) {alert('请输入用户 ID');return;}// 显示连接状态为“连接中”connectionStatus.textContent = '已连接...';connectionStatus.className = 'status';eventSource = new EventSource(`http://localhost:9999/sse/connect/${userId}`);currentUserId = userId; // 保存当前用户 IDeventSource.onopen = function() {connectionStatus.textContent = '接收成功';connectionStatus.className = 'status success';};eventSource.onmessage = function(event) {try {// 解析 JSON 消息const data = JSON.parse(event.data);const newElement = document.createElement('div');newElement.className = 'message';newElement.innerText = `用户 ${data.senderId} 接收的消息: ${data.message}`;document.getElementById('messages').appendChild(newElement);} catch (e) {console.error('消息解析错误:', e);}};eventSource.onerror = function(event) {connectionStatus.textContent = '连接失败,请检查网络或服务器';connectionStatus.className = 'status error';console.error("连接错误: ", event);eventSource.close();};}function sendMessage() {const targetUserId = document.getElementById('targetUserId').value;const message = document.getElementById('message').value;if (!targetUserId || !message) {alert('请填写目标用户 ID 和消息内容');return;}// 发送 GET 请求推送消息fetch(`http://localhost:9999/sse/push/${targetUserId}?message=${encodeURIComponent(message)}`, {method: 'GET'}).then(response => {if (response.ok) {console.log('消息发送成功');} else {console.log('消息发送失败');}}).catch(error => {console.error('发送错误:', error);alert('消息发送失败');});}</script>
</body>
</html>
后端实现:
启动类:
package io.github.deepjava;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.util.concurrent.ConcurrentHashMap;@SpringBootApplication
public class SseDemoApplication {public static void main(String[] args) {SpringApplication.run(SseDemoApplication.class);}// 注入一个全局缓存 用来保存不同用户的SSE连接信息@Bean("userSSEMap")public ConcurrentHashMap<String, SseEmitter> getUserSSEMap(){return new ConcurrentHashMap<>();}}
Controller:
package io.github.deepjava.controller;import io.github.deepjava.service.SseService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import javax.annotation.Resource;
import java.util.concurrent.ConcurrentHashMap;@RestController
@RequestMapping("/sse")
@CrossOrigin(origins = "*")
@Slf4j
public class SseController {@Resource(name = "userSSEMap")private ConcurrentHashMap<String, SseEmitter> userSSEMap;@Resourceprivate SseService sseService;// 连接方法:为用户 ID 注册 SSE 链接@GetMapping("/connect/{userId}")public SseEmitter connect(@PathVariable String userId) {SseEmitter emitter = new SseEmitter(0L); // 设置超时时间为无限大userSSEMap.put(userId, emitter);// 连接正常关闭回调 移除连接emitter.onCompletion(() -> {userSSEMap.remove(userId);log.info("连接正常关闭回调 移除连接");});// 连接超时回调 移除连接emitter.onTimeout(() -> {userSSEMap.remove(userId);log.info("连接超时回调 移除连接");});// 连接出错回调 移除连接emitter.onError((e) -> {userSSEMap.remove(userId);log.info("连接出错回调 移除连接");});log.info("连接成功!");return emitter;}// 推送方法:根据用户 ID 发送消息@GetMapping("/push/{userId}")public void push(@PathVariable String userId, @RequestParam String message) {sseService.extracted(userId, message);}}
Service:
package io.github.deepjava.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;@Service
@Slf4j
public class SseService {@Resource(name = "userSSEMap")private ConcurrentHashMap<String, SseEmitter> clients;public void extracted(String userId, String message) {SseEmitter emitter = clients.get(userId);if (emitter != null) {try {// 创建包含用户 ID 和消息内容的 JSON 对象String jsonMessage = String.format("{\"senderId\":\"%s\", \"message\":\"%s\"}", userId, message);emitter.send(jsonMessage);log.info("消息推送成功!");} catch (IOException e) {clients.remove(userId);log.info("消息推送失败!");}}}}
配置文件:application.properties
spring.application.name=sse-demo
server.port=9999
Maven的pom文件:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.1</version></parent><groupId>org.example</groupId><artifactId>SSE-Demo</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!--lombok --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency></dependencies><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties></project>
5、SSE简单实现(分布式应用Demo)
注意:
SSE 连接(如 SseEmitter)是持久化的、与具体服务器实例相关联的动态对象,无法直接存储在 Redis 等外部存储中。Redis 主要用于消息传递和共享数据,但无法直接管理活跃的连接。所以下面方案仅使用 Redis 进行消息广播。
解决方案概述
为了在分布式环境中实现 SSE,通常采用以下架构:
- 每个服务器实例维护本地的 SSE 连接:每个实例只管理与自身连接的客户端。
- 使用 Redis 进行消息广播:当需要向特定用户推送消息时,将消息发布到 Redis 频道。所有实例订阅该频道,并检查自己是否有需要向某个用户推送的连接。
- 用户与实例的映射:使用 Redis 存储用户与服务器实例的映射信息,确保消息能够被正确路由到处理该用户连接的实例。
虽然无法完全将连接信息存储在 Redis 中,但通过这种方式,可以有效地在分布式环境中管理 SSE 连接和消息推送。
这里只贴主要的后端代码:完整代码去下载仓库代码看。
SSE-Demo仓库地址
https://github.com/deepjava-gm/SSE-Demo.git
关键代码
redis配置:
// 配置redis的序列化@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {RedisTemplate<String, Object> template = new RedisTemplate<>();template.setConnectionFactory(factory);template.setKeySerializer(new StringRedisSerializer());template.setValueSerializer(new StringRedisSerializer());return template;}@Beanpublic RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);return container;}
redis监听主题:
package io.github.deepjava.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentHashMap;@RestController
@RequestMapping("/dis/sse")
@CrossOrigin(origins = "*")
@Slf4j
public class DistributedSseController {@Resource(name = "userSSEMap")private ConcurrentHashMap<String, SseEmitter> userSSEMap;@Resourceprivate RedisTemplate<String, String> redisTemplate;@Resourceprivate RedisMessageListenerContainer redisMessageListenerContainer;private final ChannelTopic topic = new ChannelTopic("sse-messages");@PostConstructpublic void init() {// 订阅 Redis 频道redisMessageListenerContainer.addMessageListener(new MessageListenerAdapter((MessageListener) (message, pattern) -> {String payload = new String(message.getBody(), StandardCharsets.UTF_8);// 假设消息的格式为 "userId:message"String[] parts = payload.split(":", 2);if (parts.length == 2) {String userId = parts[0];String userMessage = parts[1];// 发送消息给本地的 SSE 连接SseEmitter emitter = userSSEMap.get(userId);if (emitter != null) {try {String jsonMessage = String.format("{\"senderId\":\"%s\", \"message\":\"%s\"}", userId, userMessage);emitter.send(jsonMessage);} catch (IOException e) {emitter.completeWithError(e);userSSEMap.remove(userId);}}}}), topic);}// 连接方法:为用户 ID 注册 SSE 链接@GetMapping("/connect/{userId}")public SseEmitter connect(@PathVariable String userId) {SseEmitter emitter = new SseEmitter(0L); // 设置超时时间为无限大userSSEMap.put(userId, emitter);// 连接正常关闭回调 移除连接emitter.onCompletion(() -> {userSSEMap.remove(userId);log.info("连接正常关闭回调 移除连接");});// 连接超时回调 移除连接emitter.onTimeout(() -> {userSSEMap.remove(userId);log.info("连接超时回调 移除连接");});// 连接出错回调 移除连接emitter.onError((e) -> {userSSEMap.remove(userId);log.info("连接出错回调 移除连接");});log.info("连接成功!");return emitter;}@GetMapping("/push/{userId}")public void push(@PathVariable String userId, String message) {// 将消息发布到 Redis 频道redisTemplate.convertAndSend(topic.getTopic(), userId + ":" + message);}
}
方案说明
SSE 连接管理:
使用 ConcurrentHashMap<String, SseEmitter> 存储用户的连接信息,每个服务器实例只维护与自身连接的客户端。
connect 方法用于创建 SSE 连接并保存到本地缓存。
Redis 消息广播:
通过 Redis 的 发布订阅(Pub/Sub) 机制,所有实例订阅同一个频道(sse-messages)。
push 方法将消息发布到 Redis 频道,所有订阅了该频道的实例都会收到消息,并检查是否有对应的连接需要推送。
这篇关于Web服务端通过SSE推送消息给浏览器客户端的实现方案(附详细代码和仓库地址)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!