Go + Redis 实现可恢复的 LLM 流式推送：断线不丢数据的实战方案

做 LLM 流式输出的时候用户刷新一下页面流就断了后端还在跑token 白烧。本文分享一种基于 Redis Streams 的断线续传方案附完整 Go 代码。一、问题背景最近做了一个 AI 对话服务后端 GoLLM 输出通过 SSE 推给前端。上线后发现一个很头疼的问题用户刷新页面 → SSE 断了 → 后端还在跑 → token 白烧用户看不到输出 → 点重新生成 → 又烧一遍钱分布式部署下LLM worker 和 HTTP handler 不在同一台机器重连后负载均衡打到新节点找不到原来的流二、JS 生态有方案Go 没有调研了一圈JS/TS 已经有成熟方案方案说明vercel/resumable-streamVercel 官方绑定 AI SDKdurable-streamsElectricSQL 出品多语言 SDK需要跑专用 serverai-resumable-stream社区方案也是绑 AI SDK核心思路都一样chunk 存 Redis断线重连时 replay。但 Go 生态里一个能用的都没有。三、方案设计自己造了个轮子streamhub3.1 架构LLM Worker (Instance A) │ ├── Publish chunks ──→ Redis Stream持久化 │ │ │ ├──→ Consumer (Instance B) ──→ SSE │ └──→ 新连接自动 replay 历史 │ └── Listen cancel ←── Redis Pub/Sub ←── 任意节点3.2 两个 Redis 原语Redis StreamsXADD/XREAD存 chunk支持从任意位置回放Redis Pub/Sub传 cancel 信号延迟低3.3 防重复机制Generation ID作为 fencing token旧 producer 的写入会被拒绝单 Producer 注册同 session 只允许一个 producer不会重复调 LLM四、核心代码4.1 安装go get github.com/gtoxlili/streamhubv0.1.04.2 创建 Hubclient,_:rueidis.NewClient(rueidis.ClientOption{InitAddress:[]string{127.0.0.1:6379},})hub:streamhub.New(client)4.3 生产端stream,created,err:hub.Register(chat:123,func(){// 收到 cancel 信号的回调llmCancel()})if!created{return// 其他实例已经在跑了}deferstream.Close()// 可以设置 metadatastream.SetMetadata(map[string]any{model:gpt-4})fortoken:rangellmOutput{stream.Publish(token)}关键点created为false说明已有 producer不要重复生产。4.4 消费端任意实例stream:hub.Get(chat:123)ifstreamnil{return// session 不存在}chunks,unsub:stream.Subscribe(128)deferunsub()forchunk:rangechunks{// 自动先 replay 历史再无缝切 livefmt.Fprintf(w,data: %s\n\n,chunk)w.(http.Flusher).Flush()}4.5 远程取消hub.Get(chat:123).Cancel()// 通过 Redis Pub/Sub 广播producer 所在实例收到回调五、对比特性streamhubvercel/resumable-streamdurable-streams语言GoTypeScript多语言存储复用现有 RedisRedis专用 server断线 replay✅✅✅跨实例 cancel✅❌❌单 producer✅❌❌额外依赖无Vercel AI SDK需部署 server六、适用场景LLM / AI Agent 流式响应需要断线续传SSE / WebSocket 推送要求不丢数据微服务架构生产者消费者在不同实例从其他服务远程取消正在进行的生成任务七、总结核心就是一句话把流的状态从进程内存搬到 Redis让生产和消费彻底解耦。项目地址github.com/gtoxlili/streamhub目前还在早期阶段API 可能会调整。如果你也在做类似的项目欢迎提 Issue 交流。