使用 Go 实现 SSE 流式推送 + 打字机效果（模拟 Coze Chat）
使用 Go 实现 SSE 流式推送 + 打字机效果（模拟 Coze Chat）
在开发实时聊天、AI 助手或者协作应用时，我们经常需要SSE（Server-Sent Events）实现服务端向前端持续推送数据。本文将分享一个Go SSE 打字机式输出实现，并附上上游模拟示例、curl 测试和前端实时渲染示例。
功能特点
- 使用SSE推送消息流，前端无需轮询。
- 对消息进行逐字符打字机式输出，模拟 AI 打字效果。
- 支持上游 SSE 模拟，方便本地测试。
- 可轻松扩展为真实 AI 聊天接口的代理服务。
技术栈
- Go 1.21+
- CloudWeGo Hertz 作为 HTTP 框架
- resty 用于上游 SSE 请求
- SSE 流式推送（使用
hertz-contrib/sse）
完整示例代码
packagemainimport("bufio""context""encoding/json""fmt""math/rand""strings""time""github.com/cloudwego/hertz/pkg/app""github.com/cloudwego/hertz/pkg/app/server""github.com/cloudwego/hertz/pkg/common/hlog""github.com/hertz-contrib/sse""resty.dev/v3")funcmain(){h:=server.Default(server.WithHostPorts(":8380"))h.POST("/v3/chat",CozeParseThenTypeWriter)h.GET("/upstream",MockUpstreamSSE)hlog.Info("🚀 Coze SSE parse + typewriter proxy running at :8380")h.Spin()}// 核心逻辑：解析 conversation.message.delta → 打字机式输出funcCozeParseThenTypeWriter(ctx context.Context,c*app.RequestContext){// SSE Headerc.SetStatusCode(200)h:=c.Response.Header h.Set("Content-Type","text/event-stream; charset=utf-8")h.Set("Cache-Control","no-cache, no-store, must-revalidate")h.Set("Pragma","no-cache")h.Set("Connection","keep-alive")h.Set("X-Accel-Buffering","no")stream:=sse.NewStream(c)// Resty 上游请求client:=resty.New().SetTimeout(0)resp,err:=client.R().SetContext(ctx).SetDoNotParseResponse(true).SetHeader("Accept","text/event-stream").Get("http://localhost:8380/upstream")iferr!=nil||resp.RawResponse==nil||resp.RawResponse.Body==nil{hlog.Error("upstream connect failed")return}deferresp.RawResponse.Body.Close()// 打字机准备r:=rand.New(rand.NewSource(time.Now().UnixNano()))scanner:=bufio.NewScanner(resp.RawResponse.Body)varcurrentEventstring// SSE 解析循环forscanner.Scan(){select{case<-ctx.Done():hlog.Warn("client disconnected")returndefault:}line:=scanner.Text()ifline==""{currentEvent=""continue}ifstrings.HasPrefix(line,"event:"){currentEvent=strings.TrimSpace(strings.TrimPrefix(line,"event:"))continue}ifstrings.HasPrefix(line,"data:"){payload:=strings.TrimSpace(strings.TrimPrefix(line,"data:"))ifpayload=="[DONE]"{stream.Publish(&sse.Event{Data:[]byte("[DONE]")})return}ifcurrentEvent=="conversation.message.delta"{vardstruct{Contentstring`json:"content"`}iferr:=json.Unmarshal([]byte(payload),&d);err==nil{typeWriter(stream,r,d.Content)}}ifcurrentEvent=="conversation.message.completed"{stream.Publish(&sse.Event{Event:"conversation.message.completed",Data:[]byte(`{"status":"completed"}`),})}}}}// 打字机逐字符输出functypeWriter(stream*sse.Stream,r*rand.Rand,textstring){fori,ch:=rangetext{time.Sleep(getSleepDuration(r,ch))data:=map[string]any{"id":fmt.Sprintf("char_%d_%d",i,time.Now().UnixNano()%100000),"role":"assistant","type":"answer","content":string(ch),"created_at":time.Now().UnixMilli(),}b,_:=json.Marshal(data)_=stream.Publish(&sse.Event{Event:"conversation.message.delta",ID:fmt.Sprintf("char_%d",i),Data:b,})}}// 延迟策略funcgetSleepDuration(r*rand.Rand,chrune)time.Duration{switch{casech=='\n'||ch=='。'||ch=='！'||ch=='？':returntime.Duration(300+r.Intn(200))*time.Millisecondcasech=='、'||ch==' '||ch=='-'||ch==':'||ch==',':returntime.Duration(150+r.Intn(100))*time.Millisecondcasech=='#'||ch=='*'||ch=='>':returntime.Duration(200+r.Intn(150))*time.Milliseconddefault:returntime.Duration(60+r.Intn(60))*time.Millisecond}}// 上游 Mock（模拟 Coze SSE）funcMockUpstreamSSE(ctx context.Context,c*app.RequestContext){c.SetStatusCode(200)c.Header("Content-Type","text/event-stream")c.Header("Cache-Control","no-cache")c.Header("Connection","keep-alive")c.Header("X-Accel-Buffering","no")c.Flush()send:=func(event,datastring){c.Write([]byte("event: "+event+"\n"+"data: "+data+"\n\n",))c.Flush()}deltas:=[]string{"你","好","，","这","是"," Coze"," SSE"}for_,ch:=rangedeltas{send("conversation.message.delta",`{"content":"`+ch+`"}`)time.Sleep(80*time.Millisecond)}send("conversation.message.completed",`{}`)c.Write([]byte("data: [DONE]\n\n"))c.Flush()}使用方法
- 启动服务
go run main.go- 访问接口
上游 SSE 测试（浏览器可直接访问）：
http://localhost:8380/upstream打字机代理接口（POST）：
http://localhost:8380/v3/chat
使用curl测试 SSE
# 测试上游 SSEcurl-N http://localhost:8380/upstream# 测试打字机代理 SSEcurl-N -X POST http://localhost:8380/v3/chat参数说明：
-N/--no-buffer：禁用输出缓存，实时显示流式数据。-X POST：因为代理接口是 POST。
运行后，你会在终端看到类似打字机逐字符输出：
event: conversation.message.delta data: {"id":"char_0_12345","role":"assistant","type":"answer","content":"你","created_at":1700000000000} event: conversation.message.delta data: {"id":"char_1_67890","role":"assistant","type":"answer","content":"好","created_at":1700000000050} ... event: conversation.message.completed data: {"status":"completed"} data: [DONE]前端实时渲染打字机效果示例
在前端，可以使用EventSource监听 SSE 并动态显示内容：
<divid="chat"></div><script>constchatDiv=document.getElementById("chat");constevtSource=newEventSource("http://localhost:8380/v3/chat");evtSource.addEventListener("conversation.message.delta",e=>{constdata=JSON.parse(e.data);chatDiv.innerHTML+=data.content;});evtSource.addEventListener("conversation.message.completed",e=>{console.log("消息完成");});evtSource.onopen=()=>console.log("连接已打开");evtSource.onerror=()=>console.log("连接错误或关闭");</script>效果：消息逐字符显示，模拟 AI 打字机输出。
核心解析
- SSE Header 设置
h.Set("Content-Type","text/event-stream; charset=utf-8")h.Set("Cache-Control","no-cache, no-store, must-revalidate")h.Set("Connection","keep-alive")h.Set("X-Accel-Buffering","no")保证浏览器或代理实时接收流式数据。
- 打字机效果
根据字符类型不同设置不同延迟：
casech=='\n'||ch=='。'||ch=='！'||ch=='？':returntime.Duration(300+r.Intn(200))*time.Millisecond- 上游 SSE 模拟
方便本地测试，无需真实 AI 接口即可验证前端打字机效果。
总结
通过本文示例，你可以快速实现：
- Go SSE 服务端代理
- AI 聊天消息打字机式输出
- 上游 SSE 模拟
- curl 测试和前端实时渲染