Inicio Explorar Ayuda
Registro Iniciar sesión
vectorSeek
/
zsjz-ai
5
0
Fork 0
Archivos Incidencias 0 Pull Requests 0 Wiki
Árbol: 83fd98a44d
Ramas Etiquetas
zsjz-ai
/
.qoder
/
repowiki
/
zh
/
content
/
后端服务架构
/
SSE 心跳与长连接保活.md

SSE 心跳与长连接保活.md 25 KB
Histórico Raw

SSE 心跳与长连接保活

本文引用的文件

  • SseService.java
  • SseDTO.java
  • AgentChatController.java
  • AgentChatServiceImpl.java
  • application.yaml（后端）
  • application.yaml（网关）
  • sse.ts
  • chatStream.ts

目录

  1. 引言
  2. 项目结构定位
  3. 核心组件
  4. 架构总览
  5. 详细流程分析
  6. 依赖关系分析
  7. 性能与资源影响
  8. 故障排查指南
  9. 结论

引言

本项目使用两种基于 Server-Sent Events 的实时通信机制：

场景 技术路径 主要用途
Agent 流式对话 Spring WebFlux Flux<ServerSentEvent<String>>，控制器直接返回事件流 工具调用、思考过程、增量文本、完成与错误帧
清洗/治理等后台任务进度 Spring MVC SseEmitter + @Scheduled 定时心跳 数据清洗、数据治理、AI 清洗状态快照、日志推送

两者虽然都叫“SSE”，但实现层不同：Agent 对话是响应式流；清洗治理进度是传统阻塞式 SseEmitter。本文重点说明两者的心跳机制、断连重连策略、消息乱序处理，以及长连接对服务器资源的实际影响。

项目结构定位

SSE 相关代码分布在三个层次：

graph TB
    subgraph "前端"
        FE_SSE["sse.ts<br/>通用 SSE 解析器"]
        FE_Store["chatStream.ts<br/>对话流式状态机"]
    end

    subgraph "网关"
        GW_App["application.yaml<br/>连接超时、鉴权、路由"]
    end

    subgraph "后端服务"
        Controller["AgentChatController.java<br/>/chat/stream"]
        Service["AgentChatServiceImpl.java<br/>Reactor SSE 流"]
        EmitterSvc["SseService.java<br/>SseEmitter 心跳"]
        DTO["SseDTO.java<br/>清洗/治理消息模型"]
    end

    FE_SSE --> Controller
    FE_Store --> FE_SSE
    Controller --> Service
    Service --> Controller
    EmitterSvc --> DTO

图表来源

  • sse.ts:1-222
  • chatStream.ts:1-560
  • AgentChatController.java:120-170
  • AgentChatServiceImpl.java:340-620
  • SseService.java:1-172
  • SseDTO.java:1-25

章节来源

  • AgentChatController.java:120-170
  • SseService.java:1-172
  • sse.ts:1-222
  • chatStream.ts:1-560

核心组件

后端 Agent 流式对话

Agent 聊天接口由控制器暴露 /chat/stream，内容类型为 text/event-stream。服务层将底层 Agent 的事件转换为一系列 SSE 事件帧：

事件类型 含义 是否可重复
intent 意图识别结果 每轮一次
suggestions 下一步引导建议 每轮末尾可能追加
tool_call 工具调用开始 每次工具调用一次
tool_input 工具入参完整 每次工具调用一次
tool_result 工具结果 每次工具调用一次
graph_data 全量关系图数据 仅 get_case_graph 触发
thinking 模型思考过程 start/delta/end 一轮多次
token 增量回复文本 大量帧
heartbeat 服务端心跳 每 10 秒一次
done 正常结束 每轮一次
error 异常结束 每轮最多一次

章节来源

  • AgentChatController.java:120-170
  • AgentChatServiceImpl.java:479-574
  • AgentChatServiceImpl.java:591-600

后端清洗/治理 SSE 推送

SseService 维护用户到 SseEmitter 的映射，承担三类职责：

  1. 建立连接：创建带 6 小时超时的 SseEmitter，注册完成、超时、异常回调，并立即发送一条“连接成功”初始消息。
  2. 业务推送：按当前上下文或显式传入的用户 ID 推送清洗、治理、日志等进度。
  3. 定时心跳：每 5 秒遍历活跃连接发送心跳，防止代理或浏览器断开空闲连接。

该服务已改为按用户隔离连接：同一用户重复连接会断开旧连接，避免多标签页或多端同时登录时覆盖对方进度。

章节来源

  • SseService.java:1-172
  • SseDTO.java:1-25

前端 SSE 解析器

sse.ts 提供通用解析器 createParser，负责：

  • 按规范处理 \n\n / \r\n\r\n 分隔；
  • 忽略注释行；
  • 合并多行 data:；
  • 把 JSON 载荷分发给 onToken、onDone、onError、onHeartbeat 等回调；
  • 在流结束时调用 flush()，避免最后一帧丢失。

streamSse 使用原生 fetch + ReadableStream 读取流，因为普通 axios 无法消费增量数据。它还会判断 Content-Type 是否为 text/event-stream，以区分未登录时后端返回的普通 JSON 错误响应。

章节来源

  • sse.ts:1-149
  • sse.ts:150-222

前端对话状态机

chatStream.ts 是 AI 对话的流式状态管理中心，负责：

  • 乐观插入用户消息和占位助手消息；
  • 维护工具调用步骤、思考过程、执行时序链；
  • 节流富内容块解析，避免长回答造成 UI 卡顿；
  • 区分“模型活动”和“心跳活动”；
  • 处理连接中断、完成信号缺失、主动中止等生命周期。

章节来源

  • chatStream.ts:1-220
  • chatStream.ts:370-560

架构总览

sequenceDiagram
    participant User as "用户"
    participant Frontend as "前端 chatStream.ts"
    participant Parser as "前端 sse.ts"
    participant Gateway as "网关 application.yaml"
    participant Controller as "AgentChatController"
    participant Service as "AgentChatServiceImpl"
    participant Heartbeat as "心跳 Flux"

    User->>Frontend: 发送对话消息
    Frontend->>Parser: streamSse(url, params, handlers)
    Parser->>Gateway: POST /js/a/chat/stream
    Gateway->>Controller: 转发请求
    Controller->>Service: stream(req)
    Service->>Service: 会话校验、并发控制、构建 Agent
    Service->>Service: 订阅 Agent 事件
    Service->>Parser: 依次发送 tool_call/tool_result/token/thinking
    Service->>Heartbeat: 启动 10 秒心跳
    Heartbeat-->>Parser: event: heartbeat
    Service->>Parser: event: done
    Parser-->>Frontend: onDone()
    Frontend->>Frontend: 停止流式态、刷新会话元信息

图表来源

  • chatStream.ts:370-560
  • sse.ts:150-222
  • AgentChatController.java:150-170
  • AgentChatServiceImpl.java:340-620

详细流程分析

Agent 流式对话工作流

Agent 对话的核心流程如下：

flowchart TD
    Start["收到 /chat/stream 请求"] --> Validate["校验消息与会话"]
    Validate --> AcquireGate{"会话级 Semaphore 抢占成功？"}
    AcquireGate -->|否| ReturnError["返回 error 帧：会话正在回复中"]
    AcquireGate -->|是| BuildAgent["构建 RuntimeContext 与 Agent"]
    BuildAgent --> Subscribe["订阅 Agent 事件流"]
    Subscribe --> Dispatch["按事件类型转换 SSE 帧"]
    Dispatch --> ToolCall["tool_call → tool_input → tool_result"]
    Dispatch --> Thinking["thinking start → delta → end"]
    Dispatch --> Token["token 增量文本"]
    Dispatch --> Graph["graph_data（可选）"]
    Dispatch --> Done["AgentEndEvent → done"]
    Dispatch --> Error["异常 → error"]
    Dispatch --> Heartbeat["独立心跳：每 10 秒"]
    Heartbeat --> Merge["merge(body, heartbeats)"]
    Token --> Merge
    ToolCall --> Merge
    Thinking --> Merge
    Graph --> Merge
    Done --> End["释放 Semaphore、持久化结果"]
    Error --> End

图表来源

  • AgentChatServiceImpl.java:340-620

关键点：

  1. 会话级并发锁：每个 sessionId 对应一个 Semaphore(1)，防止同一会话同时发起两条流式对话。
  2. 事件顺序真实：后端按 Agent 实际执行顺序发送事件，前端用 trace 数组记录思考与工具调用的到达顺序。
  3. 超长结果截断：工具结果按工具名称限制最大字符数，避免单帧过大。
  4. 心跳独立于正文：心跳通过 Flux.interval(Duration.ofSeconds(10)) 与业务事件 Flux.merge 合并，不参与正文累积。
  5. 取消保护：用户主动中止时，若尚未收到 AgentResultEvent，会异步保存已累积的文本和工具事件。

章节来源

  • AgentChatServiceImpl.java:340-400
  • AgentChatServiceImpl.java:479-600

心跳任务与服务端定时保活

本项目有两套心跳：

心跳 位置 间隔 作用
Agent 对话心跳 AgentChatServiceImpl 中的 Flux.interval 10 秒 让前端区分“模型仍在处理”和“连接已断”
清洗/治理心跳 SseService.fixedSee 定时任务 5 秒 防止代理/Nginx/浏览器断开空闲 SSE 连接 
flowchart TD
    Scheduled["@Scheduled(fixedRate = 5000)"] --> CheckEmpty{"emitters 为空？"}
    CheckEmpty -->|是| Return["直接返回"]
    CheckEmpty -->|否| Iterate["遍历 userId → emitter"]
    Iterate --> Send["发送 SseDTO(cmd=999, message=心跳)"]
    Send --> CatchErr{"发送异常？"}
    CatchErr -->|是| Remove["移除失效连接并记录日志"]
    CatchErr -->|否| Next["下一个连接"]
    Remove --> Next
    Next --> Done["本轮心跳结束"]

图表来源

  • SseService.java:118-150

注意：调度线程池默认只有一根线程，因此配置中将其放大为 4 线程，避免 SSE 心跳、数据源回收、清洗看门狗、附件清理等定时任务互相阻塞。

章节来源

  • SseService.java:106-150
  • application.yaml（后端）:24-33

客户端断连、重连与消息乱序处理

断连检测

前端通过两个时间戳区分两类状态：

  • lastActivityAt：最近一次模型或工具事件时间；
  • lastHeartbeatAt：最近一次心跳时间。

如果长时间没有 token、tool_result、thinking 等业务活动，但有心跳，说明连接仍存活，只是模型在处理；如果心跳也停止，则判定连接可能已断。

当连接关闭但未收到 done 时，前端不会把半截内容伪装成完成，而是标记为“连接已结束但未收到完成信号”。

重连策略

当前 sse.ts 与 chatStream.ts 的实现是：单次流式请求失败后不自动重连。重连责任交给上层 UI 逻辑：用户看到错误提示后重新发送消息。这样做避免了无界重试导致的状态混乱。

消息乱序处理

SSE 本身不保证跨网络重连后的严格顺序，因此前端采用以下策略：

  1. 事件类型明确：每种事件有固定 type，例如 token、tool_call、tool_result、done。
  2. 工具调用按 toolCallId 关联：tool_call、tool_input、tool_result、graph_data 共享同一个 toolCallId。
  3. 工具输入/结果增量拼接：toolInput 和 toolResult 由前端按字符串累积，后端也对超长结果做了截断。
  4. 执行时序链：trace 数组按事件到达顺序记录“思考节点”和“工具节点”，渲染时优先使用这个顺序，而不是猜测思考与工具的对应关系。

  5. 脏读防护：清洗/治理消息在发送前补齐 caseId，前端据此丢弃其他案件的进度帧，避免上一个案件还在跑时误渲染到新案件。

    flowchart TD
Frame["收到 SSE 帧"] --> Parse["解析事件类型"]
Parse --> IsTool{"是否工具事件？"}
IsTool -->|是| MatchId["按 toolCallId 匹配已有步骤"]
MatchId --> UpdateStep["更新 input/result/status/graphPayload"]
IsTool -->|否| IsHeartbeat{"是否心跳？"}
IsHeartbeat -->|是| UpdateHB["更新 lastHeartbeatAt"]
IsHeartbeat -->|否| AppendText["token 追加 content"]
AppendText --> Throttle["节流重扫富内容块"]
UpdateStep --> Trace["追加 trace 节点"]
UpdateHB --> Trace
Throttle --> Trace
Trace --> State["更新本地 ChatMessage 状态"]

图表来源

  • chatStream.ts:394-560
  • SseDTO.java:1-25
  • SseService.java:82-104

章节来源

  • chatStream.ts:370-560
  • sse.ts:1-149
  • SseService.java:60-104

SSE 与 WebSocket 的选型原因

本项目选择 SSE 而非 WebSocket，主要基于以下实际场景：

维度 本项目现状 说明
方向 主要是服务端单向推送到前端 Agent 对话、清洗治理进度都是服务端输出
协议复杂度 希望保持简单 SSE 基于 HTTP，天然支持鉴权、缓存、代理、Nginx 透传
双向命令 不需要复杂的双向二进制协议 控制面走普通 REST，数据面走 SSE
实现成本 两套后端技术栈并存 Reactor SSE 用于 Agent；SseEmitter 用于传统任务进度
浏览器兼容 Electron + 现代浏览器 原生 EventSource 或 fetch + ReadableStream 均可

WebSocket 更适合需要频繁双向小消息、低延迟指令通道的场景，例如实时多人协作编辑、游戏同步。而本项目的 Agent 对话本质上是“服务端逐步输出结果”，清洗治理进度也是“服务端周期性推送”，SSE 更贴合需求。

章节来源

  • AgentChatController.java:150-170
  • SseService.java:1-25

依赖关系分析

classDiagram
    class AgentChatController {
        +stream(req) Flux
        +sendMessage(dto) MessageVO
        +listMessages(sessionId) List
    }

    class AgentChatServiceImpl {
        -streamLocks : ConcurrentHashMap
        -stream(req) Flux
        -doStream(...) Flux
        -doStreamLocked(...) Flux
        +stream(ChatRequestDTO) Flux
    }

    class SseService {
        -emitters : Map
        +see() SseEmitter
        +sendSee(dto) void
        +fixedSee() void
        +closeSee(userId) void
    }

    class SseDTO {
        +cmd : int
        +process : int
        +message : String
        +seeData : SeeDataDTO
        +caseId : Long
    }

    class ChatStreamStore {
        +send(content) Promise
        +stop() void
    }

    class SseParser {
        +push(chunk) void
        +flush() void
    }

    AgentChatController --> AgentChatServiceImpl : "调用"
    AgentChatServiceImpl --> AgentChatController : "返回 ServerSentEvent"
    SseService --> SseDTO : "序列化消息"
    ChatStreamStore --> SseParser : "使用"

图表来源

  • AgentChatController.java:120-170
  • AgentChatServiceImpl.java:340-620
  • SseService.java:1-172
  • SseDTO.java:1-25
  • sse.ts:1-222
  • chatStream.ts:1-220

性能与资源影响

长连接对服务器的影响

风险 现状 建议
连接数量增长 SseService.emitters 按用户保存 SseEmitter 增加连接监控指标，设置单用户连接上限
心跳 CPU 消耗 每 5 秒遍历所有连接 连接数很大时改用批量或分段广播
调度线程阻塞 默认单线程调度器承载多个定时任务 已配置 4 线程，继续观察调度队列时长
SSE 响应内存 后端聚合工具结果、图谱、思考文本 保持超长结果截断，避免单帧过大
前端渲染压力 token 增量高频触发富内容扫描 已使用 100ms 节流，必要时可调整阈值
网关超时 网关后端连接超时 3 秒，注释指出不设全局响应超时 确保 Nginx/反向代理的 SSE 超时远大于 Agent 最长推理时间

连接池控制

当前 SseService 使用 ConcurrentHashMap<Long, SseEmitter> 直接保存连接，没有连接池或大小限制。建议：

  1. 增加最大连接数阈值；
  2. 对单用户连接数做限制（当前已允许同用户新连接踢旧连接）；
  3. 定期扫描长期无活动的连接；
  4. 对连接建立、关闭、异常、心跳失败打点统计。

超时清理

  • SseEmitter 默认超时 6 小时；
  • 连接完成、超时、异常都会从 map 中移除；
  • 心跳发送失败也会移除对应连接；
  • Agent 对话心跳随 Flux 生命周期自动终止。

背压处理

  • Agent 对话使用 Reactor Flux，天然具备背压能力；
  • 工具结果超过长度限制会被截断；
  • 前端对富内容块解析加了节流定时器；
  • 清洗治理进度是逐条推送，不存在批量堆积问题。

异常恢复策略

当前实现偏向“错误即终止”：

  • 连接异常 → 关闭连接；
  • 心跳失败 → 移除连接；
  • 前端连接中断且未收到 done → 标记中断，不自动重连；
  • 会话级并发冲突 → 直接返回错误帧。

改进方向：

  • 前端可增加指数退避重连；
  • 后端可记录连接失败率，辅助扩缩容；
  • 对长时间无输出的 Agent 对话增加应用层超时；
  • 对清洗治理任务增加断线后查询最终状态的能力。

章节来源

  • SseService.java:24-172
  • AgentChatServiceImpl.java:340-620
  • application.yaml（后端）:24-33
  • application.yaml（网关）:1-25
  • chatStream.ts:394-560

故障排查指南

查看服务端日志

关注以下日志关键字：

日志关键词 含义 排查方向
sse 连接已关闭 SSE 连接正常关闭 确认是否为用户主动关闭或页面卸载
sse 连接超时 SseEmitter 达到 6 小时超时 检查是否存在僵尸连接
sse 连接异常 发送或连接生命周期出现异常 检查网络、代理、防火墙
sse 初始消息发送失败 刚建立连接就发送失败 检查客户端是否立刻关闭连接
sse 发送失败，关闭连接 业务消息发送失败 检查连接是否已经半关闭
sse 心跳失败 定时心跳写入失败 检查连接是否已失效
Chat stream error Agent 流式对话异常 结合 sessionKey 查看下游模型或工具错误

章节来源

  • SseService.java:46-150
  • AgentChatServiceImpl.java:345-367

检查心跳间隔

  • Agent 对话心跳：10 秒；
  • 清洗/治理心跳：5 秒；
  • 前端心跳时间戳字段：lastHeartbeatAt；
  • 前端活动时间戳字段：lastActivityAt。

如果前端长时间显示“连接正常但无输出”，应检查是否有 heartbeat 事件；如果没有，说明底层连接已被代理或网关断开。

章节来源

  • AgentChatServiceImpl.java:591-600
  • SseService.java:118-150
  • chatStream.ts:394-410

确认防火墙与 Nginx 是否支持 SSE

SSE 要求反向代理：

  1. 保留 Content-Type: text/event-stream；
  2. 不过早关闭连接；
  3. 不缓冲整个响应；
  4. 不拦截或改写 event: / data: 帧；
  5. 设置足够长的超时，尤其是 Agent 对话可能持续数十秒甚至更长；
  6. 网关 connect-timeout 不应被误解为 SSE 响应超时。

当前网关配置中已明确注释：网关后端连接超时为 3 秒，但没有设置全局响应超时，原因是 SSE 研判会长时间保持连接。部署时应特别注意 Nginx 的 proxy_read_timeout 和负载均衡器的连接超时。

章节来源

  • application.yaml（网关）:1-25
  • sse.ts:150-222

常见现象对照表

现象 可能原因 建议操作
前端一直等待，无 token 也无心跳 连接在代理层被提前关闭 检查 Nginx、防火墙、负载均衡超时
有心跳但无业务输出 模型或工具仍在处理 等待；检查后端日志中是否仍有工具调用
偶尔丢最后一段文本 流结束帧未冲刷 确认前端调用了 flush()
工具结果渲染异常 JSON 被外层包裹或过长 检查归一化和截断逻辑
切换案件后看到上一个案件的进度 缺少 caseId 过滤 检查 withCaseId 补发逻辑
同一账号多标签页进度互相覆盖 同用户复用连接键 这是设计行为，与“同账号不允许多端登录”一致

章节来源

  • SseService.java:12-23
  • SseService.java:82-104
  • sse.ts:1-149
  • chatStream.ts:520-560

结论

本项目的 SSE 心跳与长连接保活分为两层：

  1. Agent 流式对话使用 Reactor SSE，心跳间隔 10 秒，主要用于让前端感知“模型仍在处理”，并通过事件类型、toolCallId、trace 时序链维持消息语义一致性。
  2. 清洗/治理进度使用 SseEmitter，心跳间隔 5 秒，主要用于防止中间网络设备和浏览器断开空闲连接。

当前实现的优点是简单可靠、前后端职责清晰；缺点是连接管理偏手动、没有显式连接池、重连策略较弱。后续优化应优先考虑连接监控、连接上限、指数退避重连、应用层超时和背压指标采集。对于生产环境，尤其要验证反向代理和负载均衡器对 text/event-stream 的支持，否则即使后端心跳正常，前端也可能因中间层超时而断连。