♥️作者：小宋1021 🤵♂️个人主页：小宋1021主页 ♥️坚持分析平时学习到的项目以及学习到的软件开发知识，和大家一起努力呀！！！ 🎈🎈加油！ 加油！ 加油！ 加油 🎈欢迎评论 💬点赞👍🏻 收藏 📂加关注+！
目录
一、业务背景：为什么我们需要“后端主动推送”
二、技术选型：SSE、WebSocket、轮询怎么选
三、SSE 核心机制速览（结合 Spring）
四、我们这次的架构设计
五、前端相关展示
页面展示
前端代码
六、后端相关展示
SSE 订阅入口Controller
WarnSseService相关代码
七、联调测试步骤（实战版）
八、常见问题与排查清单
【写在前面】 这篇文章的目标不是再讲一遍“什么是 SSE”的基础定义，而是基于一个真实业务场景，把“为什么选 SSE、怎么设计、怎么编码、怎么测试、怎么上线”讲透。场景很典型：后端持续消费 Kafka 告警流，需要按卫星维度把告警实时推送给前端页面，供值班人员快速感知与处置。
SSE相关介绍:Web实时通信的学习之旅：SSE（Server-Sent Events）的技术详解及简单示例演示-CSDN博客
一、业务背景：为什么我们需要“后端主动推送”
在故障诊断类系统里，告警是持续到来的流式数据，而不是用户手动触发的一次性查询。传统轮询模式（前端每隔 N 秒请求一次接口）有几个明显问题：
1）实时性差 轮询周期是 5 秒，就意味着最坏情况下会有接近 5 秒的可见延迟。对于告警场景，这个延迟是不可接受的。
2）无效请求多 没有新告警时，前端也在持续打空请求，浪费后端资源与网络带宽。
3）并发压力高 页面数量一多，固定频率轮询会带来可观 QPS，而且这些 QPS 的业务价值不高。
我们需要的是： – 新告警一到，后端立刻把消息推到前端； – 前端尽量少做复杂协议处理； – 不引入过重的中间层与改造成本。
这正是 SSE 的适用区间。
二、技术选型：SSE、WebSocket、轮询怎么选
先给结论：当前场景“服务端单向推送告警给前端”，SSE 是最合适的平衡点。
1）轮询 – 优点：最简单，任何后端都能做。 – 缺点：实时性受轮询周期限制，请求冗余高。
2）WebSocket – 优点：全双工，能力最强。 – 缺点：协议与网关配置复杂度更高，前端改造也更多。
3）SSE（Server-Sent Events） – 优点：基于 HTTP，浏览器原生 EventSource 支持，自动重连，后端 Spring MVC 可直接用 SseEmitter。 – 缺点：天然单向（服务端 -> 客户端），不适合强交互双向通信。
我们的需求是单向实时告警推送，所以 SSE 是“足够好且实现快”的解。
三、SSE 核心机制速览（结合 Spring）
SSE 的通信模型可以概括为一句话： “前端发起一次 GET 长连接，后端在连接存活期间持续写事件帧，前端按事件名监听消息。”在 Spring MVC 中，核心类是： – SseEmitter：表示一条 SSE 连接。
常见代码动作： – 创建连接：new SseEmitter(timeout) – 发送事件：emitter.send(SseEmitter.event().name("alarm").data(payload)) – 注册回调：onCompletion / onTimeout / onError
这些回调非常关键，因为它们决定了连接回收是否干净，直接关系内存与句柄稳定性
四、我们这次的架构设计
目标链路： Kafka -> KafkaProcess 告警处理 -> SSE 服务按卫星路由 -> 前端 EventSource 页面。
### 4.1 分层职责
1）KafkaProcess – 消费告警消息； – 进行已有业务处理（落库、野值判断、实时分组诊断等）； – 新增一步：构造 SSE 消息并推送。
2）WarnSseService – 管理订阅连接池； – 按卫星定向推送； – 支持 ALL 全量订阅； – 清理失效连接。
3）WarnSseController – 暴露订阅接口： – /warn/sse/subscribe/{satelliteCode} – /warn/sse/subscribe
4）WarnSseMessage – 统一推送消息结构，前端按固定字段消费。
### 4.2 路由策略
我们没有做“一锅端广播”，而是做了两层路由： – 第一层：卫星订阅（例如 BD001） – 第二层：ALL 订阅（用于总览大屏）
推送时： – 先推给指定卫星； – 再推给 ALL。
这种设计兼顾精确订阅和全量监控，且实现成本低。
五、前端相关展示
页面展示
这是一个简单的连接示例，可以修改后台链接以及订阅方式(单一订阅、全部订阅)，当写好连接地址之后点击开始订阅会显示连接成功字样，然后我们发送kafka后就可以在这里接到kafka的消息。
前端代码
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>SSE 告警订阅示例</title>
<style>
:root {
–bg: #f6f8fb;
–panel: #ffffff;
–text: #1f2937;
–muted: #6b7280;
–accent: #0b57d0;
–danger: #b42318;
–ok: #137333;
–border: #e5e7eb;
}
body { margin: 0; background: linear-gradient(180deg, #eef3ff, var(–bg)); font-family: "Microsoft YaHei", "PingFang SC", sans-serif; color: var(–text); }
.wrap { max-width: 980px; margin: 24px auto; padding: 0 16px; }
.card { background: var(–panel); border: 1px solid var(–border); border-radius: 12px; padding: 16px; box-shadow: 0 6px 20px rgba(0, 0, 0, 0.04); }
.title { font-size: 20px; font-weight: 700; margin: 0 0 12px; }
.row { display: flex; gap: 10px; flex-wrap: wrap; align-items: center; margin-bottom: 10px; }
input[type="text"] { height: 34px; padding: 0 10px; border: 1px solid var(–border); border-radius: 8px; min-width: 220px; }
button { height: 34px; border: 0; border-radius: 8px; padding: 0 14px; cursor: pointer; }
.btn-primary { background: var(–accent); color: #fff; }
.btn-gray { background: #e5e7eb; color: #111827; }
.status { font-size: 13px; color: var(–muted); }
.status .ok { color: var(–ok); }
.status .err { color: var(–danger); }
.table-wrap { margin-top: 14px; overflow: auto; }
table { width: 100%; border-collapse: collapse; font-size: 13px; }
th, td { border-bottom: 1px solid var(–border); text-align: left; padding: 8px 6px; white-space: nowrap; }
th { color: #374151; background: #f9fafb; position: sticky; top: 0; }
.log { margin-top: 12px; background: #0f172a; color: #cbd5e1; border-radius: 8px; padding: 10px; height: 180px; overflow: auto; font-size: 12px; }
</style>
</head>
<body>
<div class="wrap">
<div class="card">
<h1 class="title">SSE 告警订阅示例</h1>
<div class="row"><label>Backend Base URL:</label><input id="baseUrl" type="text" value="http://localhost:8001" /></div>
<div class="row"><label>Satellite Code:</label><input id="satelliteCode" type="text" value="BD001" /><label><input id="allFlag" type="checkbox" /> Subscribe ALL</label></div>
<div class="row"><button id="btnConnect" class="btn-primary">开始订阅</button><button id="btnClose" class="btn-gray">停止订阅</button><button id="btnClear" class="btn-gray">清空数据</button></div>
<div class="status" id="status">状态：未连接</div>
<div class="table-wrap"><table><thead><tr><th>time</th><th>type</th><th>satellite</th><th>tmCode</th><th>tmName</th><th>warnValue</th><th>warnMessage</th><th>outlier</th><th>skipDiagnosis</th></tr></thead><tbody id="tbody"></tbody></table></div>
<div class="log" id="log"></div>
</div>
</div>
<script src="./app.js"></script>
</body>
</html>
把这串代码放到.html文件里打开即可
前端核心是 EventSource：
const es = new EventSource('/admin/warn/sse/subscribe/BD001?clientId=xxx'); es.addEventListener('connected', …) es.addEventListener('alarm', e => JSON.parse(e.data))
关键点： 1）SSE 订阅必须是 GET 不能用 POST 建立 SSE。
2）EventSource 自动重连 网络抖动时会重连，但你仍要做 UI 状态提示。
3）连接关闭 页面销毁要 es.close()，避免后台长期占连接。
我们还给了一个可直接运行的示例页面： – frontend-sse-demo/index.html – frontend-sse-demo/app.js
可作为联调基线。
六、后端相关展示
SSE 订阅入口Controller
/**
* SSE 订阅入口。
*/
@RestController
@RequestMapping("/warn/sse")
public class WarnSseController {
@Resource
private WarnSseService warnSseService;
@GetMapping(value = "/subscribe/{satelliteCode}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter subscribeBySatellite(@PathVariable String satelliteCode,
@RequestParam(value = "clientId", required = false) String clientId) {
// 订阅某一颗卫星的报警流
return warnSseService.subscribe(satelliteCode, clientId);
}
@GetMapping(value = "/subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter subscribeAll(@RequestParam(value = "clientId", required = false) String clientId) {
// 订阅全部卫星报警流
return warnSseService.subscribe("ALL", clientId);
}
}
WarnSseService相关代码
/**
* SSE 连接管理：
* 1) 维护“卫星 -> 客户端 -> 连接”映射
* 2) 支持按卫星推送和全量推送（ALL）
*/
@Service
public class WarnSseService {
private static final Logger log = LoggerFactory.getLogger(WarnSseService.class);
private static final String ALL = "ALL";
private final ConcurrentHashMap<String, ConcurrentHashMap<String, SseEmitter>> emitterBySatellite = new ConcurrentHashMap<>();
public SseEmitter subscribe(String satelliteCode, String clientId) {
// 未传卫星时归类到 ALL，表示订阅全部卫星
String satKey = StringUtils.hasText(satelliteCode) ? satelliteCode : ALL;
// 前端未传 clientId 时生成一个，便于连接回收
String clientKey = StringUtils.hasText(clientId) ? clientId : UUID.randomUUID().toString();
SseEmitter emitter = new SseEmitter(0L);
emitterBySatellite.computeIfAbsent(satKey, key -> new ConcurrentHashMap<>()).put(clientKey, emitter);
emitter.onCompletion(() -> remove(satKey, clientKey));
emitter.onTimeout(() -> remove(satKey, clientKey));
emitter.onError(ex -> remove(satKey, clientKey));
try {
emitter.send(SseEmitter.event().name("connected").id(clientKey).data("ok"));
} catch (IOException ex) {
remove(satKey, clientKey);
emitter.completeWithError(ex);
}
return emitter;
}
public void pushWarn(String satelliteCode, Object payload) {
if (!StringUtils.hasText(satelliteCode)) {
return;
}
// 先推给该卫星订阅者，再推给全量订阅者
sendToSatellite(satelliteCode, payload);
sendToSatellite(ALL, payload);
}
private void sendToSatellite(String satelliteCode, Object payload) {
ConcurrentHashMap<String, SseEmitter> emitters = emitterBySatellite.get(satelliteCode);
if (emitters == null || emitters.isEmpty()) {
return;
}
List<String> deadClients = new ArrayList<>();
emitters.forEach((clientId, emitter) -> {
try {
emitter.send(SseEmitter.event().name("alarm").data(payload));
} catch (Exception ex) {
// 连接失效时先记录，循环结束后统一移除
deadClients.add(clientId);
}
});
for (String clientId : deadClients) {
remove(satelliteCode, clientId);
}
}
private void remove(String satelliteCode, String clientId) {
ConcurrentHashMap<String, SseEmitter> emitters = emitterBySatellite.get(satelliteCode);
if (emitters == null) {
return;
}
emitters.remove(clientId);
if (emitters.isEmpty()) {
emitterBySatellite.remove(satelliteCode);
}
log.debug("sse client removed, satelliteCode={}, clientId={}", satelliteCode, clientId);
}
}
在需要推送的地方调用pushWarn即可
/**
* 时间戳转换成时间格式
*/
private void publishWarnToSse(String alarmResultType,
String kafkaId,
WarnEntity warnEntity,
boolean outlierFlag,
boolean skipDiagnosis) {
// 将当前报警按卫星维度推送到 SSE
if (warnEntity == null || !StringUtils.hasText(warnEntity.getSatelliteCode())) {
return;
}
WarnSseMessage message = WarnSseMessage.builder()
.alarmResultType(alarmResultType)
.kafkaId(kafkaId)
.satelliteCode(warnEntity.getSatelliteCode())
.tmCode(warnEntity.getTmCode())
.tmName(warnEntity.getTmName())
.warnValue(warnEntity.getBjz())
.warnMessage(warnEntity.getBjxx())
.warnTime(warnEntity.getBjKssj())
.recoverTime(warnEntity.getBjhfsj())
.endTime(warnEntity.getBjqrsj())
.alarmLevel(warnEntity.getBjjb())
.severityLevel(warnEntity.getSeveritylevel())
.outlier(outlierFlag)
.skipDiagnosis(skipDiagnosis)
.build();
warnSseService.pushWarn(warnEntity.getSatelliteCode(), message);
}
七、联调测试步骤（实战版）
### 步骤 1：确认访问路径
本项目 admin 服务有 context-path：/admin 所以地址形如： http://localhost:8001/admin/warn/sse/subscribe/BD001?clientId=test001
### 步骤 2：确认认证策略
如果 Shiro 对该路径未放行，需要 token 或配置 anon。 否则前端会表现为 sse error。
### 步骤 3：建立订阅
浏览器控制台建 EventSource，观察是否收到 connected 事件。
### 步骤 4：投递告警
向 Kafka 投递 NEW_ALARM/UPP_MULTIPLE/END_ALARM 任一消息。 注意 satCode 要满足系统 FILTER_SATELLITE 过滤规则。
### 步骤 5：校验页面与日志
– 页面是否收到 alarm 事件； – 字段是否完整（satelliteCode、tmCode、warnTime 等）； – 后端是否有发送失败回收日志。
八、常见问题与排查清单
### 1）前端一直 sse error 优先排查： – URL 是否缺少 /admin 前缀； – 是否被鉴权拦截（401）； – 反向代理是否支持流式响应； – 后端是否真的启动在该端口。
### 2）后端看起来推送了，但前端收不到 检查： – 订阅的 satelliteCode 与消息里的 satelliteCode 是否一致； – 是否订阅了 ALL； – 前端是否监听了正确事件名（alarm）。
### 3）连接越来越多，内存上涨 检查： – onCompletion/onTimeout/onError 是否都注册； – 前端页面关闭是否调用 close； – 是否存在异常路径没有 remove。
### 4）消息顺序问题 SSE 本身是单连接有序，但多线程发送时业务上可能“近似乱序”。 如需强顺序，可以： – 在 payload 增加业务时间戳和序号； – 前端做按时间戳整理展示。
