JAVA通过SSE实现消息推送

devtools/2025/3/11 5:27:09/

JAVA通过SSE实现消息推送

  • 1.什么是SSE?
  • 2.SSE技术的基本原理
  • 3.SSE和Socket的区别
  • 4.编写SSE服务,来进行创建链接和发送消息
  • 5.前端实现消息监听

1.什么是SSE?

SSE(Server-Sent Events)是一种用于实现服务器主动向客户端推送数据的技术,也被称为“事件流”(Event Stream)。它基于 HTTP 协议,利用了其长连接特性,在客户端与服务器之间建立一条持久化连接,并通过这条连接实现服务器向客户端的实时数据推送。

2.SSE技术的基本原理

  • 客户端向服务器发送一个GET请求,带有指定的header,表示可以接收事件流类型,并禁用任何的事件缓存。
  • 服务器返回一个响应,带有指定的header,表示事件的媒体类型和编码,以及使用分块传输编码(chunked)来流式传输动态生成的内容。
  • 服务器在有数据更新时,向客户端发送一个或多个名称:值字段组成的事件,由单个换行符分隔。事件之间由两个换行符分隔。服务器可以发送事件数据、事件类型、事件ID和重试时间等字段。
  • 客户端使用EventSource接口来创建一个对象,打开连接,并订阅onopen、onmessage和onerror等事件处理程序来处理连接状态和接收消息。
  • 客户端可以使用GET查询参数来传递数据给服务器,也可以使用close方法来关闭连接。

3.SSE和Socket的区别

SSE(Server-Sent Events)和 WebSocket 都是实现服务器向客户端实时推送数据的技术,但它们在某些方面还是有一定的区别。

技术实现
SSE 基于 HTTP 协议,利用了其长连接特性,通过浏览器向服务器发送一个 HTTP 请求,建立一条持久化的连接。而 WebSocket 则是通过特殊的升级协议(HTTP/1.1 Upgrade 或者 HTTP/2)建立新的 TCP 连接,与传统 HTTP 连接不同。

数据格式
SSE 可以传输文本和二进制格式的数据,但只支持单向数据流,即只能由服务器向客户端推送数据。WebSocket 支持双向数据流,客户端和服务器可以互相发送消息,并且没有消息大小限制。

连接状态
SSE 的连接状态仅有三种==:已连接、连接中、已断开==。连接状态是由浏览器自动维护的,客户端无法手动关闭或重新打开连接。而 WebSocket 连接的状态更灵活,可以手动打开、关闭、重连等。

兼容性
SSE 是标准的 Web API,可以在大部分现代浏览器和移动设备上使用。但如果需要兼容老版本的浏览器(如 IE6/7/8),则需要使用 polyfill 库进行兼容。而 WebSocket 在一些老版本 Android 手机上可能存在兼容性问题,需要使用一些特殊的 API 进行处理。

安全性
SSE 的实现比较简单,都是基于 HTTP 协议的,与普通的 Web 应用没有太大差异,因此风险相对较低。WebSocket 则需要通过额外的安全措施(如 SSL/TLS 加密)来确保数据传输的安全性,避免被窃听和篡改,否则可能会带来安全隐患。

总体来说,SSE 和 WebSocket 都有各自的优缺点,适用于不同的场景和需求。如果只需要服务器向客户端单向推送数据,并且应用在前端的浏览器环境中,则 SSE 是一个更加轻量级、易于实现和维护的选择。而如果需要双向传输数据、支持自定义协议、或者在更加复杂的网络环境中应用,则 WebSocket 可能更加适合。

SSE适用于场景
SSE适用场景是指服务器向客户端实时推送数据的场景,例如:

  • 股票价格更新:服务器可以根据股市的变化,实时地将股票价格推送给客户端,让客户端能够及时了解股票的走势和行情。
  • 新闻实时推送:服务器可以根据新闻的更新,实时地将新闻内容或标题推送给客户端,让客户端能够及时了解最新的新闻动态和信息。
  • 在线聊天:服务器可以根据用户的发送,实时地将聊天消息推送给客户端,让客户端能够及时收到和回复消息。
  • 实时监控:服务器可以根据设备的状态,实时地将监控数据或报警信息推送给客户端,让客户端能够及时了解设备的运行情况和异常情况。

SSE适用场景的特点是:

  • 数据更新频繁:服务器需要不断地将最新的数据推送给客户端,保持数据的实时性和准确性。
  • 低延迟:服务器需要尽快地将数据推送给客户端,避免数据的延迟和过期。
  • 单向通信:服务器只需要向客户端推送数据,而不需要接收客户端的数据。

4.编写SSE服务,来进行创建链接和发送消息

Service:

java">package com.zillion.aggregate.app.controller;import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;@Slf4j
@Service
public class SSEService {private static final Map<String,SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();public SseEmitter crateSse(String uid) {SseEmitter sseEmitter = new SseEmitter(0L);sseEmitter.onCompletion(() -> {log.info("[{}]结束链接" , uid);sseEmitterMap.remove(uid);});sseEmitter.onTimeout(() -> {log.info("[{}]链接超时",uid);});sseEmitter.onError(throwable -> {try{log.info("[{}]链接异常,{}",uid,throwable.toString());sseEmitter.send(SseEmitter.event().id(uid).name("发生异常").data("发生异常请重试").reconnectTime(3000));sseEmitterMap.put(uid,sseEmitter);}catch (IOException e){e.printStackTrace();}});try{sseEmitter.send(SseEmitter.event().reconnectTime(5000));}catch (IOException e){e.printStackTrace();}sseEmitterMap.put(uid,sseEmitter);log.info("[{}]创建sse连接成功!",uid);return sseEmitter;}public boolean sendMessage(String uid,String messageId,String message){if(StringUtils.isEmpty(message)){log.info("[{}]参数异常,msg为空",uid);return false;}SseEmitter sseEmitter = sseEmitterMap.get(uid);if(sseEmitter == null){log.info("[{}]sse连接不存在",uid);return  false;}try{sseEmitter.send(SseEmitter.event().id(messageId).reconnectTime(60000).data(message));log.info("用户{},消息ID:{},推送成功:{}",uid,messageId,message);return true;}catch (IOException e){sseEmitterMap.remove(uid);log.info("用户{},消息ID:{},消息推送失败:{}",uid,messageId,message);sseEmitter.complete();return false;}}public void closeSse(String uid){if(sseEmitterMap.containsKey(uid)){SseEmitter sseEmitter = sseEmitterMap.get(uid);sseEmitter.complete();sseEmitterMap.remove(uid);}else {log.info("用户{}连接已关闭",uid);}}}

Controller:

java">package com.zillion.aggregate.app.controller;import cn.hutool.core.util.IdUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;@Controller
@RequestMapping("/aggregate/api/pay")
public class TestController {private static final Map<String,Boolean> SEND_MAP = new ConcurrentHashMap<>();@Autowiredprivate SSEService sseService;@GetMapping("createSse")@CrossOriginpublic SseEmitter createSse(String uid){return sseService.crateSse(uid);}@GetMapping("/sendMsg")@ResponseBody@CrossOriginpublic SseEmitter sendMsg(@RequestParam("uid") String uid) throws InterruptedException {SseEmitter sseEmitter = sseService.crateSse(uid);if (SEND_MAP.get(uid)==null ||  !SEND_MAP.get(uid)){new Thread(()->{int i=0;while (true){try {i++;String message = "uid:"+uid+" number:"+i+" message:"+IdUtil.fastUUID().replace("-", "");sseService.sendMessage(uid,"消息"+i,message);SEND_MAP.put(uid,true);Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();closeSse(uid);}}}).start();}return sseEmitter;}@GetMapping("closeSse")@CrossOriginpublic void closeSse(String uid){sseService.closeSse(uid);}
}

5.前端实现消息监听

java"><!doctype html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1"><title>SSE消息推送监听</title>
</head>
<body><div id="conMsg"></div>
<script>let uid = 1;let chat = document.getElementById("conMsg");if(window.EventSource){var eventSource = new EventSource(`http://localhost:9001/aggregate/aggregate/api/pay/sendMsg?interfaceId=CEDB297CECCC9DCBAD348204ACDD5BAD&uid=${uid}`);eventSource.onopen = ()=>{console.log("链接成功");}eventSource.onmessage = (ev)=>{if(ev.data){chat.innerHTML += ev.data+"<br>";}}eventSource.onerror = ()=>{console.log("sse链接失败")}}else{alert("当前浏览器不支持sse")}
</script>
</body>
</html>

http://www.ppmy.cn/devtools/166193.html

相关文章

蓝桥杯 字符串拼接【省模拟赛】

问题描述 给定四个字符串 a,b,c,da,b,c,d&#xff0c;请将这四个字符串按照任意顺序依次连接拼成一个字符串。 请问拼成的字符串字典序最小是多少&#xff1f; 输入格式 输入四行&#xff0c;每行包含一个字符串。 输出格式 输出一行包含一个字符串&#xff0c;表示答案。 样例…

React基础之组件

在React中一个组件就是首字母大写的函数&#xff0c;内部存放了组件的逻辑和视图UI&#xff0c;渲染组件只需要把组件当作标签书写即可 //定义组件 // function Button(){ // return <button>click me&#xff01;</button> // } //也可以使用箭头函数来定义 co…

Gazebo不报错但是没有机器人模型

现象是&#xff0c;gazebo能打开&#xff0c;有世界模型&#xff0c;但是没有机器人模型&#xff1b;排查过不是模型文件的问题&#xff0c;因为啥模型都有这样的现象。 这种情况可以参考以下解决办法&#xff1a; &#xff08;1&#xff09;看看catkin_ws中有没有gazebo_ros…

【AI】AI开源IDE:CLine源码分析报告

1. 源码位置&#xff1a; CLine 是一个开源的 VSCode 插件&#xff0c;其完整源码托管在 GitHub 的 cline/cline 仓库中。这个仓库包含 CLine 的核心逻辑&#xff08;TypeScript 编写&#xff09;&#xff0c;包括与 LLM 的对话控制、工具调用接口&#xff0c;以及 VSCode 插件…

Android 检查更新

首先服务类 public class UpdateService extends Service {private static final String NOTIFY_CHANNEL_ID "com.jianke.api.UpdateService";public static final String BROADCAST_UPDATE_VERSION_AUTH_INSTALL_APK "BROADCAST_UPDATE_VERSION_AUTH_INSTAL…

【定制开发】碰一碰发视频系统定制开发,支持OEM

在短视频营销爆发的2025年&#xff0c;"碰一碰发视频"技术已成为实体商家引流标配。某连锁餐饮品牌通过定制化开发&#xff0c;单月视频发布量突破10万条&#xff0c;获客成本降低80%&#xff01;本文将深入解析该系统的技术架构与开发要点&#xff0c;助你快速搭建高…

linux的top指令解析

这张图片显示的是 Linux 系统中 top 命令的输出&#xff0c;它用于实时监控系统的资源使用情况&#xff0c;包括 CPU、内存和进程等信息。以下是对图片中几个关键参数的解释&#xff1a; 系统负载 (Load Average) 1分钟、5分钟、15分钟负载平均值&#xff1a;这三个数值表示系…

vue实现一个pdf在线预览,pdf选择文本并提取复制文字触发弹窗效果

[TOC] 一、文件预览 1、安装依赖包 这里安装了disjs-dist2.16版本&#xff0c;安装过程中报错缺少worker-loader npm i pdfjs-dist2.16.105 worker-loader3.0.8 2、模板部分 <template><div id"pdf-view"><canvas v-for"page in pdfPages&qu…