实习生日常：大模型流式响应（SSE）在前端断网重连时的后端状态恢复- 惊觉

实习生日常：大模型流式响应（SSE）在前端断网重连时的后端状态恢复

封面信息图

在开发基于 Server-Sent Events（SSE）的大模型流式对话与工单辅助生成功能时，前端移动端或弱网环境下的网络抖动（WiFi 切换 4G/5G、偶发丢包）是一项极具挑战的工程问题。

在传统的单次 HTTP 请求中，如果网络断开，客户端只需重新发起一次请求。
但在大模型长达 15~30 秒的流式推送过程中：

如果大模型已经生成了 200 个字，此时前端网络闪断了 2 秒；
浏览器底层的 EventSource 会自动发起重连；
如果后端没有做好状态暂存与游标机制，后端要么把整个长耗时任务从头重新生成一遍（浪费大量 Token 费用），要么直接返回 500 报错，导致用户屏幕上的文字彻底中断！

上周我主导设计并落地了一套基于 Redis 环形缓冲区（Ring Buffer） + SSE Last-Event-ID 游标的流式断网无缝续传方案。

今天我把这套高可用流式状态恢复机制完整记录下来。

SSE 协议的天然断线重连标准：`Last-Event-ID`

很多同学只知道 SSE 是通过 text/event-stream 推送数据，却忽略了 W3C 标准中专门为断网重连设计的字段：

后端下发事件 ID：
后端在推送每一个数据块（Chunk）时，可以附带一个自增的 id: 101\n；
客户端自动携带上次 ID 重连：
当网络断开并自动重连时，浏览器会在 HTTP 请求头中自动带上 Last-Event-ID: 101；
后端补发历史差量数据：
后端收到 Last-Event-ID 后，只需从缓存中把 id > 101 的未收到数据块依次补发给客户端，即可实现无缝续传！

graph TD
    A[客户端] -->|1. 发起流式请求 SSE| B[后端服务]
    B -->|2. 推送 Chunk 1 (id: 1)| A
    B -->|3. 推送 Chunk 2 (id: 2)| A
    A -.->|4. 发生 2 秒网络闪断| B
    B -->|5. 模型继续生成 Chunk 3, 4 并暂存 Redis| C[Redis 环形缓冲区]
    A -->|6. 携带 Last-Event-ID: 2 发起重连| B
    B -->|7. 从 Redis 读取 id > 2 的差量数据| C
    B -->|8. 补发 Chunk 3, 4 并继续后续流式推送| A

核心架构实现：基于 Redis List 的会话级流式暂存器

在调用大模型生成流式 Token 时，后端不能仅仅把 Token 往客户端打，必须同步写入 Redis 临时环形队列（TTL 5 分钟）：

@Service
@Slf4j
public class ResilientSseStreamService {

    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private AiInferenceClient aiInferenceClient;

    public SseEmitter handleStreamWithResume(String sessionId, String prompt, @Nullable String lastEventIdStr) {
        SseEmitter emitter = new SseEmitter(60_000L); // 60 秒超时
        String redisKey = "sse:stream:" + sessionId;

        // 1. 如果携带了 Last-Event-ID，说明是断网重连，先补发历史漏掉的数据块
        int lastSeenId = (lastEventIdStr != null && !lastEventIdStr.isBlank()) ? Integer.parseInt(lastEventIdStr) : 0;
        if (lastSeenId > 0) {
            replayMissedEvents(redisKey, lastSeenId, emitter);
        }

        // 2. 如果是全新请求 (lastSeenId == 0)，开启大模型异步生成任务
        if (lastSeenId == 0) {
            startLlmGenerationTask(redisKey, sessionId, prompt, emitter);
        } else {
            // 如果是重连请求，大模型任务通常已经在后台运行中，监听 Redis 即可
            attachToExistingStream(redisKey, lastSeenId, emitter);
        }

        return emitter;
    }

    private void replayMissedEvents(String redisKey, int lastSeenId, SseEmitter emitter) {
        // 从 Redis 列表中按索引拉取漏掉的事件
        List<String> allEvents = redisTemplate.opsForList().range(redisKey, lastSeenId, -1);
        if (allEvents != null) {
            for (int i = 0; i < allEvents.size(); i++) {
                int eventId = lastSeenId + 1 + i;
                try {
                    emitter.send(SseEmitter.event().id(String.valueOf(eventId)).data(allEvents.get(i)));
                } catch (IOException e) {
                    log.warn("补发事件时再次断开连接: sessionId={}", redisKey);
                    break;
                }
            }
        }
    }

    private void startLlmGenerationTask(String redisKey, String sessionId, String prompt, SseEmitter emitter) {
        AtomicInteger eventSeq = new AtomicInteger(0);

        aiInferenceClient.streamChat(prompt)
                .doOnNext(chunk -> {
                    int eventId = eventSeq.incrementAndGet();
                    // 同步落入 Redis 缓存供重连使用
                    redisTemplate.opsForList().rightPush(redisKey, chunk);
                    redisTemplate.expire(redisKey, 5, TimeUnit.MINUTES);

                    try {
                        emitter.send(SseEmitter.event().id(String.valueOf(eventId)).data(chunk));
                    } catch (Exception e) {
                        // 客户端断开不抛出致命异常，允许大模型在后台继续把答案生成完毕落入 Redis
                        log.warn("客户端连接临时中断，保持后台生成: sessionId={}", sessionId);
                    }
                })
                .doOnComplete(() -> {
                    try {
                        emitter.send(SseEmitter.event().name("done").data("[DONE]"));
                        emitter.complete();
                    } catch (Exception ignored) {}
                })
                .subscribe();
    }

    private void attachToExistingStream(String redisKey, int lastSeenId, SseEmitter emitter) {
        // 监听后续流式结束（略）
    }
}

生产环境的三大防护策略

解耦客户端连接与大模型后台生成：
当客户端因弱网断开时，坚决不要立即取消（Cancel）底层的大模型推理任务。让模型继续在后台把完整内容推完并暂存 Redis，这样用户下一次重连时可以瞬间秒级拉取完整答案，极大提升体验并避免 Token 浪费；
设置合理的 Redis 暂存 TTL（5 分钟）：
单次会话的数据块在生成结束后 5 分钟自动过期释放，防止 Redis 内存暴涨；
前端防抖与指数退避重连：
前端在捕捉到 EventSource.onerror 时，不要以 10ms 极短间隔无脑狂刷重连，应配合 500ms、1000ms、2000ms 的指数退避，给后端留出缓冲区写入时间。

实习生的实战总结

在 AI 时代做工程，稳定性的内涵发生了深刻变化。
不仅要保证后端接口自身不宕机，更要深入网络传输协议底层，为弱网、长连接和非确定性 I/O 做好全链路的容错与状态恢复设计。

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/cannonjinx/article/details/164757701

实习生日常：大模型流式响应（SSE）在前端断网重连时的后端状态恢复