GoFr 如何用 response.Stream 把任务进度与日志实时推流给客户端【免费下载链接】gofrAn opinionated GoLang framework for accelerated microservice development. Built in support for databases and observability.项目地址: https://gitcode.com/GitHub_Trending/go/gofr当一个 HTTP 接口需要随时间逐步返回结果——长任务的进度百分比、日志行的实时尾巴、语言模型的输出 token——GoFr 的默认行为handler 返回单个值、框架在返回时一次性发送就不够用了。GoFr 为此提供response.Streamhandler 返回它之后框架在每产生一个值时立刻写给客户端并 flush而不是把整个响应缓存在内存里。本文基于 官方流式响应文档 和 response.Stream 源码演示如何写一个把任务进度和日志实时推给客户端的 handler并说明格式选择、心跳、断连处理和验证方式。实现 Streamer流的值从哪来response.Stream的Source字段要求实现response.Streamer接口——一个被框架逐个拉取pull的迭代器type Streamer interface { Next() (any, bool) // 返回下一个值和 true结束时返回零值和 false Err() error // 流结束后若有终止错误在这里返回 Close() error // 释放资源并解除对阻塞中的 Next 的阻塞 }源码注释 明确了并发契约responder 会在一个 goroutine 上调Next并可能在另一个 goroutine 上调Close例如客户端断连时。如果你的Close无法让阻塞中的Next返回断连时生产者 goroutine 会泄漏。所以Close必须能立刻解除阻塞的Next。写一个进度推流 handler以一个长任务为例jobProgress依次发出 25、50、75、100 的进度值后结束以下代码来自官方文档// jobProgress emits 25, 50, 75, 100 and then ends. type jobProgress struct{ pct int } func (p *jobProgress) Next() (any, bool) { if p.pct 100 { return nil, false } p.pct 25 return map[string]any{progress: p.pct}, true } func (p *jobProgress) Err() error { return nil } func (p *jobProgress) Close() error { return nil } func trackJob(c *gofr.Context) (any, error) { return response.Stream{Source: jobProgress{}}, nil }把它注册为路由后每个值都会被 JSON 编码并立刻 flush客户端在生产出来时就收到{progress:25}、{progress:50}……而不是在任务结束时一次性收到全部。选择线格式SSE 还是 NDJSONStream.Format字段选择线格式零值即 Server-Sent Eventsreturn response.Stream{Source: src, Format: response.NDJSON}, nilresponse.SSE默认——每个值写成data: json\n\n流正常结束时以data: [DONE]终止。适合浏览器端的EventSource。response.NDJSON——每行一个 JSON 值。适合按行读取流式响应的程序化客户端。Stream.Heartbeat设置 SSE 流空闲时的 keep-alive 发送间隔零值使用默认间隔实现 中为 15 秒。心跳的作用是在没有数据流动时也能探测到已经掉线的客户端。NDJSON 下心跳不往线路上写数据字节只走一次写路径做探测。背压与客户端断连背压值通过无缓冲 channel 按需拉取慢客户端会自然拖慢生产者而不是让服务端把整条流缓存在内存里。客户端断连客户端离开时GoFr 会拆除流并调用你的Source.Close()。Close必须解除阻塞中的Next例如关闭它读取的 channel否则生产者 goroutine 会泄漏。客户端断连后产生的值可能被直接丢弃。对于日志尾巴这类从 channel 读取的行流Close通常就是关闭源 channel// 示例从 channel 读日志行的 StreamerClose 关闭 channel 以解除 Next 的阻塞 type logTail struct { lines -chan string done chan struct{} } func (l *logTail) Next() (any, bool) { line, ok : -l.lines return line, ok } func (l *logTail) Err() error { return nil } func (l *logTail) Close() error { close(l.done); return nil }注意一个边界响应状态在第一个值产生之前就已提交为200 OK所以一个立刻失败的 Source 返回的仍是200加一个错误帧handler 随Stream一起返回的 error 会被忽略。验证流式输出项目测试断言了确切的线上格式可作为对照的示例结果stream_test.goSSE 正常结束的响应体文档示例data: a data: b data: [DONE]NDJSON 响应的Content-Type为application/x-ndjsonSSE 则为text/event-stream并带Cache-Control: no-cache和Connection: keep-alive头。源出错时例如某个值无法 JSON 编码SSE 流会追加错误帧示例结果event: error data: {error:stream value could not be encoded}客户端侧的验证方式取决于格式SSE 用浏览器的EventSource或任何 SSE 客户端按事件读取确认每个事件是独立 JSON 且收到[DONE]NDJSON 用程序化客户端逐行读取 JSON并容忍心跳产生的空行。可选同一机制推 LLM tokenresponse.Stream是传输原语不绑定生产者。推语言模型输出是同一类型的另一种用法ctx.LLM().Stream(...)返回一个response.Streamerhandler 直接把它作为Source返回即可例如return response.Stream{Source: s}, nil。完整可运行示例见 examples/using-ai/main.go 的POST /stream路由AI 侧细节见 Calling LLMs 文档。写完后检查你的Streamer.Close是否真的能解除阻塞中的Next这是断连场景下唯一容易踩的坑断连时框架只负责调用Close生产者 goroutine 能不能退出取决于你自己的实现。【免费下载链接】gofrAn opinionated GoLang framework for accelerated microservice development. Built in support for databases and observability.项目地址: https://gitcode.com/GitHub_Trending/go/gofr创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考