1. 从一次实时日志推送需求说起为什么最终选了这套组合去年底接手一个内部运维平台的重构核心诉求很明确后端有一批常驻任务在跑任务执行过程中会持续产生日志和状态变更前端页面需要近乎实时地看到这些输出而不是让用户手动点刷新。一开始团队里有人提议直接上WebSocket理由是全双工、成熟、资料多。但真正把链路捋清楚之后我们最终选了另一条路Gin做HTTP入口 gRPC双向流做内部消息通道 SSE做浏览器侧推送。这个组合乍看有点绕为什么不用一套WebSocket从头打到尾原因在于我们的系统架构本身是微服务化的任务执行器是独立的gRPC服务网关层是Gin写的HTTP服务。如果前端直接连WebSocket到网关网关再转WebSocket到执行器那中间要维护两套长连接协议状态同步和断线重连逻辑会翻倍。而gRPC双向流天然适合服务端持续推、客户端也能随时发指令的场景SSE又天然适合浏览器只需要单向接收的场景。各用各的长处反而比强行统一更省心。这篇文章就把这套链路的完整实现拆开讲清楚gRPC双向流怎么建、Gin怎么把流数据转成SSE、前端EventSource怎么接、中间有哪些坑。适合已经写过Gin、对gRPC有基本概念、想搞实时推送但不想被WebSocket复杂度拖住的中高级开发者。如果你只是想要一个最简单的推送Demo那SSE单独用就够了不必上gRPC但只要你的后端是分布式、任务执行和接入分离的这套模式的价值就会立刻体现出来。先说清楚一个容易混淆的点SSE不是WebSocket的替代品它俩解决的不是同一类问题。SSE是单向的、基于HTTP的、文本优先的WebSocket是双向的、独立协议的、二进制友好的。我们的场景里浏览器到服务端的指令很少启动任务、停止任务用普通POST就够了真正高频的是服务端到浏览器的推送这正是SSE的主场。而gRPC双向流解决的是服务与服务之间的实时通信两者职责清晰不重叠。2. 三层链路的职责划分与数据流向2.1 为什么不让Gin直接调执行器而要多一层gRPC流很多人第一反应是Gin收到前端请求后直接去调用执行器的HTTP接口拿数据不就行了问题在于持续推送这个动作。如果Gin用HTTP轮询执行器那本质上还是轮询延迟和资源消耗都下不来。如果执行器暴露一个HTTP的流式接口chunkedGin再去转发那其实也能做但执行器那边就得自己实现一套流式HTTP的服务端逻辑还要处理背压、心跳、断流检测工作量并不比gRPC小。gRPC双向流的优势在于协议层已经帮你把流控、心跳、连接状态、错误码都定义好了。执行器只需要实现一个StreamTaskLog(stream TaskService_StreamTaskLogServer)方法在方法里循环stream.Send()就行客户端那边stream.Recv()循环接收。连接断了gRPC会返回明确的错误码重连逻辑也好写。而且gRPC的protobuf序列化在日志这种结构化数据上比JSON更省带宽字段增减也更安全。这里有个关键设计决策双向流里的双向到底用在哪。我们的用法是客户端Gin侧建立流之后先发一个TaskRequest告诉执行器我要订阅哪个任务的日志然后执行器开始持续Send日志。同时客户端还可以在同一个流里继续Send控制指令比如暂停输出调整日志级别。如果只是单向订阅其实用server streaming就够了但双向流给了我们后续扩展控制能力的空间不用改协议。2.2 SSE在浏览器侧的真实工作方式SSE的本质是客户端发一个普通的HTTP GET请求服务端返回的响应头里带上Content-Type: text/event-stream然后保持连接不关闭持续往响应体里写数据。浏览器端的EventSource对象会自动解析这些数据按照data:、event:、id:、retry:这几个字段的格式触发对应的事件回调。它和普通HTTP请求最大的区别是响应体是流式的、永不结束的直到主动关闭或出错。这意味着Gin这边不能用普通的c.JSON()或c.String()必须拿到http.ResponseWriter手动设置header然后Flush()把数据推出去。很多人第一次写SSE会发现数据明明写了但前端收不到十有八九是忘了Flush或者被中间的反向代理缓冲了。数据流向完整走一遍是这样的浏览器EventSource发起GET请求到Gin的/api/task/stream接口Gin的handler里建立一个到执行器的gRPC双向流发送订阅请求然后进入一个循环从gRPC流Recv到一条日志格式化成SSE的data:格式写入ResponseWriter并Flush。浏览器收到后触发onmessage前端把日志追加到页面上。用户点停止任务时前端发一个普通POSTGin通过另一个gRPC调用或复用同一个流通知执行器停止。2.3 三层各自的边界与不该做的事把职责划清楚能避免很多返工。执行器层只负责产生日志和状态不关心谁来消费、怎么消费它只管往流里Send。Gin网关层负责协议转换和连接管理它要处理SSE的连接生命周期、gRPC流的建立与关闭、错误透传但不应该做业务逻辑判断。浏览器层只负责展示和发送用户指令不关心数据是从gRPC还是别的地方来的。一个常见的越界是在Gin层做日志的过滤、聚合、格式化。这些如果放在执行器做能减少传输量如果放在前端做能减轻网关压力。我的建议是原始日志原样透传格式化交给前端因为前端最清楚当前页面要展示什么。网关层只做必要的协议转换保持薄。3. gRPC双向流的服务端与客户端实现细节3.1 proto定义里那些容易写错的字段先看proto。双向流的关键字是stream两边都加service TaskService { rpc StreamTaskLog(stream TaskRequest) returns (stream TaskLogResponse); } message TaskRequest { string task_id 1; string action 2; // subscribe / pause / resume / stop string payload 3; } message TaskLogResponse { string task_id 1; int64 timestamp 2; string level 3; string content 4; bool finished 5; }这里有几个我踩过的点。第一action字段用字符串而不是enum虽然不够严谨但在快速迭代期改起来方便不用重新生成代码。第二finished字段很重要它告诉客户端这个任务的日志已经结束了可以关闭流了否则客户端会一直等。第三timestamp用int64存毫秒别用string前端处理起来方便。生成代码时注意双向流的服务端接口签名是StreamTaskLog(TaskService_StreamTaskLogServer) error客户端是StreamTaskLog(ctx) (TaskService_StreamTaskLogClient, error)。服务端拿到的stream对象既能Recv也能Send客户端拿到的也是。别搞混了。3.2 服务端在循环里同时处理接收和发送服务端的核心逻辑是一个循环但这里有个陷阱如果只在循环里Recv就没法主动Send如果只在循环里Send就没法响应客户端的控制指令。标准做法是开两个goroutine一个专门Recv一个专门Send用channel通信。func (s *taskServer) StreamTaskLog(stream pb.TaskService_StreamTaskLogServer) error { ctx : stream.Context() var taskID string logCh : make(chan *pb.TaskLogResponse, 100) ctrlCh : make(chan *pb.TaskRequest, 10) // 接收协程 go func() { for { req, err : stream.Recv() if err ! nil { close(ctrlCh) return } ctrlCh - req } }() // 发送协程 go func() { for log : range logCh { if err : stream.Send(log); err ! nil { return } } }() // 主循环处理控制指令 for req : range ctrlCh { switch req.Action { case subscribe: taskID req.TaskId go s.produceLogs(ctx, taskID, logCh) case stop: // 通知任务停止 close(logCh) return nil } } return nil }logCh的缓冲大小设成100是个经验值太小容易阻塞生产日志的goroutine太大内存占用高。produceLogs里就是实际读日志、构造TaskLogResponse、往channel里塞。注意ctx.Done()要监听客户端断开时能及时清理。3.3 客户端Gin侧如何建立并持有这条流Gin侧建立流的代码不复杂但流的生命周期管理是重点。每个SSE连接对应一条gRPC流连接关闭时必须把流也关掉否则执行器那边会一直挂着。func (h *Handler) StreamTask(c *gin.Context) { taskID : c.Query(task_id) ctx, cancel : context.WithCancel(c.Request.Context()) defer cancel() stream, err : h.client.StreamTaskLog(ctx) if err ! nil { c.JSON(500, gin.H{error: err.Error()}) return } // 发送订阅请求 stream.Send(pb.TaskRequest{TaskId: taskID, Action: subscribe}) // 设置SSE响应头 c.Writer.Header().Set(Content-Type, text/event-stream) c.Writer.Header().Set(Cache-Control, no-cache) c.Writer.Header().Set(Connection, keep-alive) c.Writer.Header().Set(X-Accel-Buffering, no) c.Writer.Flush() for { resp, err : stream.Recv() if err ! nil { break } data, _ : json.Marshal(resp) fmt.Fprintf(c.Writer, data: %s\n\n, data) c.Writer.Flush() if resp.Finished { break } } }X-Accel-Buffering: no这个头是给Nginx看的不加的话Nginx会缓冲响应前端要等缓冲区满了才收到数据实时性就没了。c.Request.Context()绑定到请求上浏览器断开时context自动cancelgRPC流也会跟着结束这是最省心的清理方式。4. SSE在Gin里的落地与前端EventSource对接4.1 Gin写SSE必须绕过的几个坑Gin的c.Writer本质是http.ResponseWriter的包装写SSE时不要用c.JSON或c.String它们会设置Content-Length或做额外处理破坏流式语义。直接用fmt.Fprintf(c.Writer, ...)然后c.Writer.Flush()。第二个坑是中间件。如果你的Gin路由上挂了日志中间件、gzip中间件它们可能会读取或缓冲响应体。gzip中间件尤其致命它会把流式数据压缩后攒着发。解决办法是给SSE路由单独分组不挂这些中间件或者用c.Writer的原始接口绕过。第三个坑是超时。Gin默认的http.Server有ReadTimeout和WriteTimeout如果设了WriteTimeoutSSE连接会在超时后被强制关闭。要么把这两个值设大要么用http.Server的IdleTimeout配合SSE自己的心跳。我的做法是SSE路由所在的server不设WriteTimeout靠心跳和context来控制。4.2 心跳、重连与事件ID的正确用法SSE连接长时间没数据中间的负载均衡或代理可能会把它掐掉。所以必须发心跳一般15到30秒一次发一个注释行:\n\n或者一个空的data:都行。注释行更省事前端不会触发onmessage。ticker : time.NewTicker(20 * time.Second) defer ticker.Stop() go func() { for range ticker.C { fmt.Fprintf(c.Writer, : heartbeat\n\n) c.Writer.Flush() } }()EventSource自带重连默认3秒一次。服务端可以通过retry:字段调整这个间隔。id:字段配合Last-Event-ID请求头能实现断点续传但我们的场景里日志是连续的断了就断了重连后重新订阅即可所以没用到id。如果你的场景需要精确续传那就要在执行器侧维护一个可回溯的日志缓冲复杂度会上升不少。4.3 前端EventSource的封装与状态管理前端用EventSource很简单但要处理好重连时的重复订阅和页面卸载时的关闭。class TaskLogStream { constructor(taskId) { this.taskId taskId; this.es null; } start(onLog) { this.es new EventSource(/api/task/stream?task_id${this.taskId}); this.es.onmessage (e) { const data JSON.parse(e.data); onLog(data); if (data.finished) this.close(); }; this.es.onerror () { // EventSource会自动重连这里只做提示 console.warn(SSE连接中断正在重连...); }; } close() { if (this.es) { this.es.close(); this.es null; } } }页面卸载时一定要调close()否则浏览器会保持连接服务端那边gRPC流也不会断。Vue或React里在onUnmounted/useEffect的清理函数里调。另外注意EventSource只支持GET所以task_id只能放query里不能放body。5. 实测中暴露的问题与排查链路5.1 日志延迟三秒才到一次完整的排查过程上线测试时发现日志有大约3秒的延迟不是实时的。排查链路是这样的先在前端onmessage里打时间戳确认是接收端慢还是发送端慢。结果发现前端收到的时间戳和服务端Send的时间戳差了3秒说明问题在中间。第一步怀疑gRPC在客户端Recv后打日志发现Recv返回的时间和Send时间几乎一致gRPC没问题。第二步怀疑Gin的Flush检查代码发现Flush确实调了。第三步怀疑Nginx查配置发现proxy_buffering是默认的on。虽然加了X-Accel-Buffering: no但那个头只对部分版本生效最稳妥的是在Nginx的location里显式写proxy_buffering off;。改完之后延迟降到毫秒级。这个坑的教训是SSE的实时性依赖链路上每一环都不缓冲。Gin的Flush、Nginx的proxy_buffering、甚至某些云厂商的负载均衡都可能有缓冲。排查时从后往前逐段打时间戳很快就能定位。5.2 连接数上涨后执行器内存飙升压测时发现并发SSE连接到200左右执行器内存开始飙升。原因是每个gRPC流对应一个produceLogsgoroutine每个goroutine里又有一个logCh缓冲100条200个连接就是200个channel加上goroutine本身的栈内存就上去了。优化方案有两个一是把logCh缓冲降到20日志生产速度远快于消费时自然会阻塞但gRPC的流控会处理背压不会丢数据二是多个订阅同一任务的连接共享一个日志生产goroutine用发布订阅模式一个任务只读一次日志分发给所有订阅者。第二个方案改动大但收益明显我们最终采用了它内存降了60%以上。5.3 客户端断开后执行器还在跑context传递的坑有个bug是浏览器关了页面执行器那边的任务日志还在往一个已经没人收的流里写。原因是produceLogs里没有监听ctx.Done()。gRPC的stream.Context()在客户端断开时会cancel但如果你在produceLogs里用的是自己的context.Background()就收不到这个信号。修复很简单把stream.Context()一路传下去在produceLogs的循环里加select { case -ctx.Done(): return; default: }。这个坑很隐蔽因为功能上没问题只是资源泄漏压测时才暴露。6. 性能调优与生产环境的加固建议6.1 背压处理别让慢消费者拖垮执行器SSE的消费者浏览器可能因为网络慢或页面卡顿而消费缓慢如果执行器不管不顾地SendgRPC的流控窗口满了之后Send会阻塞进而阻塞日志生产。正确的做法是给每个订阅者一个有界缓冲缓冲满了就丢弃或降级。日志这种数据丢几条通常可以接受但状态变更不能丢。所以我们的策略是日志类消息缓冲满了直接丢并计数状态类消息用阻塞发送保证不丢。这个区分要在消息类型上做标记执行器侧根据类型走不同的发送路径。6.2 多实例部署时SSE连接的粘性问题Gin网关如果部署多个实例浏览器的一个SSE连接只会落到其中一个实例上这本身没问题。但如果执行器也是多实例且任务和执行器有绑定关系那网关在建立gRPC流时要确保连到正确的执行器实例。我们的做法是网关维护一个任务到执行器地址的映射从注册中心拿建立流时按task_id路由。如果任务可以在任意执行器上跑那就简单了随便连一个都行。另一个问题是网关实例重启时所有SSE连接会断前端EventSource会自动重连重连后可能落到另一个网关实例这时要能重新建立到执行器的流。因为我们的订阅是无状态的订阅请求里带task_id所以重连后重新订阅即可不需要额外的会话保持。6.3 监控指标该埋在哪几个点生产环境必须埋的指标当前活跃SSE连接数Gin侧、当前活跃gRPC流数执行器侧、每条流的消息发送速率和丢弃数、从日志产生到前端接收的端到端延迟。前三个用Prometheus的Gauge和Counter就能搞定端到端延迟需要在前端上报或者用一条特殊的心跳日志来测量。我特别建议埋丢弃数这个指标因为背压丢消息是静默的不埋指标根本发现不了。我们有一次线上日志偶尔缺行查了半天才发现是某个执行器实例的缓冲经常满丢弃数一直在涨。7. 几个我实际踩过之后才明白的经验第一个经验SSE的data字段里不能有裸换行。如果你的日志内容里本身有\n直接塞进data:会导致SSE解析错乱因为SSE用换行分隔字段。正确做法是把内容做JSON序列化JSON里的换行会被转义成\n就不会破坏SSE格式了。这也是为什么我前面一直用json.Marshal再写data:。第二个经验gRPC流的Send和Recv不要放在同一个goroutine里。虽然文档说可以但实际用起来如果Send阻塞了Recv就没机会执行客户端的控制指令就响应不了。开两个goroutine是最稳的代价只是多一个channel。第三个经验别用context.WithTimeout包SSE的context。SSE连接本来就是要长期保持的加超时会导致连接被定时切断。要用context.WithCancel靠浏览器断开或主动关闭来cancel。第四个经验测试时一定要模拟慢消费者。用curl加--limit-rate限速或者在前端onmessage里加sleep看看执行器和网关在消费者慢的时候表现如何。很多背压问题只有在这种场景下才暴露。这套组合拳打下来我们的实时日志推送稳定跑了大半年单网关实例支撑上千并发SSE连接没问题端到端延迟稳定在50毫秒以内。如果你也在做类似的实时推送不妨按这个思路试试尤其是后端已经是微服务架构的情况gRPC双向流加SSE的分工真的比硬上WebSocket清爽很多。