SSE实现消息实时推送，前端渐进式学习、实践，真香

一、SSE概念

SSE(Server Sent Event)，直译为服务器发送事件，顾名思义，也就是客户端可以获取到服务器发送的事件。我们常见的 http 交互方式是客户端发起请求，服务端响应，然后一次请求完毕；但是在 sse 的场景下，客户端发起请求，连接不停保持，服务端有数据就可以返回数据给客户端，这个返回可以是多次间隔的方式
二、SSE应用场景

在web端消息推送功能中，由于传统的HTTP协议是由客户端主动发起请求，服务端才会响应。基本的ajax轮询技能便是云云。而在SSE中，浏览发送一个请求给服务端，通过响应头中的Content-Type:text/event-stream等向客户端声名这是一个长连接，发送的是流数据，这样客户端就不会关闭连接，不停等待服务端发送数据。
如果服务器返回的数据中包含了事件标识符，浏览器会记载最后一次吸收的事件的标识符。如果与服务器的连接中断，当浏览器再次举行连接时，会通过头来声明最后一次吸收的事件的标识符。服务器端可以通过浏览器发送的事件标识符来确定从哪个事件来继续连接
三、前端使用方法、题目

1、get方式

使用eventsource完成get请求
缺点：客户端无法通过一个get请求完成数据传递
参考文档：
EventSource - Web API 接口参考 | MDN
实现流程：

• 后端提供了两个接口，一个是：post,用以完成前端信息的传递，我这边是做大语言模型的，所以包罗了模型必要参数、题目等；二、get接口，完成流式输出的接口，设置相应的具名事件、请求头等
  
• 前端通过调用post接口拿到本次会话id，将id携带在get请求里，完成信息传递
  
• 前端处理SSE流式返回
  

代码实现：

1. const eventSourceRef = useRef<any>(null)
2. const contact = async (messageData: any) => {
3.   eventSourceRef.current = new EventSource(
4.     `${API_BASE}/v1/model/stream?id=${id}`,
5.   )
6.   if (!eventSourceRef.current) return
7.   // 监听 SSE 事件，因为后端定义了具名事件，所以这儿要用addEventListener监听，而不是onmessage
8.   eventSourceRef.current.addEventListener('add', function (e: any) {
9.         // 处理数据展示
10.   })
11.   eventSourceRef.current.addEventListener('finish', function (e: any) {
12.    // 结束标识finish
13.     eventSourceRef.current.close() // 关闭连接
14.   })
15.   eventSourceRef.current.addEventListener('error', function (e: any) {
16.     if (e.status === 401) {
17.       // 用户登录状态失效处理
18.     }
19.     // error报错处理
20.     console.log('Error occurred:', e)
21.     // 关闭连接
22.     eventSourceRef.current.close()
23.   })
24. }

复制代码

2、post方式

使用fetch-event-source完成连接，仅需一个接口，支持添加请求头
缺点：在浏览器返回的text/eventstream里看不到具体返回，无法举行预览
参考文档：
@microsoft/fetch-event-source
实现流程：

• 后端提供一个接口，支持前端传参、流式返回
  
• 前端通过fetch-event-source，完成传参、请求头添加等
  
• 处理返回数据
  

具体实现：

1.   const eventSourceRef = useRef<any>(null)
2.   const [abortController, setAbortController] = useState(new AbortController())
3.   // 通信事件
4.   const contact = async (messageData: any) => {
5.     messageData = { ...messageData, do_stream: modelArg.do_stream } // 请求参数
6.     receivedDataRef.current = ''
7.     const token: string = getLocal('AUTHCODE') || ''
8.     fetchEventSource(`${MAAS_API_BASE}/v1/model_api/invoke`, {
9.       method: 'POST',
10.       // 添加请求头
11.       headers: {
12.         Authorization: token,
13.         'Content-Type': 'application/json',
14.       },
15.        // 传参必须保证是json
16.       body: JSON.stringify(messageData),
17.       // abortController.signal 提供了一个信号对象给 fetchEventSource 函数。
18.       // 如果在任何时候你想取消正在进行的 fetch 操作，你可以调用
19.       // abortController.abort()。这会发出关联任务的信号，你可以使用
20.       // AbortController 的信号来检查异步操作是否已被取消。
21.       signal: abortController.signal,
22.       openWhenHidden: true, // 切换标签页时连接不关闭
23.       async onopen(resp) {
24.         // 处理登录失效
25.         if (resp.status === 401) {
26.           message.warning('登录过期')
27.           return
28.         }
29.       },
30.       onmessage(msg: any) {
31.         const eventType = msg.event // 监听event的具名事件
32.         switch (eventType) {
33.           case 'add':
34.             // 流式输出事件，add每次会返回具体字符，前端负责拼接展示
35.             break
36.           case 'finish':
37.             setStatu('finish') // 结束标识
38.             break
39.           case 'error':
40.             if (msg.status === 401) {
41.                message.warning('登录过期')
42.             }
43.             console.log('Error occurred:', e)
44.             break
45.         }
46.       },
47.       onerror(err) {
48.         throw err // 连接遇到http错误时，如跨域等，必须要throw才能停止，不然会一直重连
49.       },
50.       onclose() {},
51.     })
52.   }
54.   // 终止连接方法，比如在切换模型时，你可能有必要终止上一次连接来避免问答串联
55.   const closeSSE = () => {
56.     abortController.abort()
57.     setAbortController(new AbortController())
58.   }

复制代码

3、一种接口同时兼容流式/非流式

同上post方法

1.     fetchEventSource(sseUrl, {
2.       method: 'POST',
3.       headers,
4.       signal: abortController.signal,
5.       body: JSON.stringify(customInferData),
6.       openWhenHidden: true,
7.     /**
8.     *在onopen阶段处理
9.     第一步：判断resp.headers.get('content-type')，如果不包含text/event-stream，
10.     则代表非流式
11.     第二步：需要在onopen阶段处理非流式返回，即json返回,读取json返回并渲染，注意异常也要处理
12.     第三步：
13.      */
14.       async onopen(resp) {
15.         const contentype = resp.headers.get('content-type') || ''
16.         console.log('contentype =>', contentype)
17.         console.log('resp.ok =>', resp.ok)
18.         if (resp.ok && !contentype.includes('text/event-stream')) {
19.           // 读取json数据
20.           const responseData = await resp.json()
21.           if (responseData.code !== 0) {
22.              // 报错处理+关闭连接
23.           } else {
24.             //处理数据渲染+关闭连接
25.             
26.             stopSession()
27.           }
28.         } else if (resp.status === 401) {
29.           message.warning('登录过期')
30.           // 报错处理+关闭连接
31.           stopSession()
32.         
33.         }
34.       },
35.       onmessage(msg: any) {
36.         const eventType = msg.event
37.         const messages: any = cloneDeep(chatState.sessionMessages)
38.         let lastMessage: any = messages[messages.length - 1] || {}
40.         switch (eventType) {
41.           case 'add':
42.             lastMessage = {
43.               ...lastMessage,
44.               text: `${lastMessage.text}${msg.data || ' '}`,
45.               loading: false,
46.             }
47.             messages.splice(messages.length - 1, 1, lastMessage)
48.             chatAction.updateSessionMessages(messages)
49.             break
50.           case 'finish':
51.             console.log('finish lastMessage =>', lastMessage)
52.             chatAction.updateSessionStatu(SessionStatuTypes.ready)
53.             chatAction.updateContext(msg.data)
54.             break
55.           case 'info':
56.             {
57.               const messages: any = cloneDeep(chatState.sessionMessages)
58.               let lastMessage: any = messages[messages.length - 1] || {}
59.               lastMessage = {
60.                 referenceDocs: JSON.parse(msg.data).reference_by_docs,
61.                 ...lastMessage,
62.               }
63.               messages.splice(messages.length - 1, 1, lastMessage)
64.               chatAction.updateSessionMessages(messages)
65.             }
66.             break
67.           case 'error':
68.             if (msg.status === 401) {
69.               chatAction.updateSessionStatu(SessionStatuTypes.ready)
70.             } else {
71.               errorItemFn(msg?.msg || msg?.data || '抱歉，暂无法回答问题')
72.             }
73.             break
74.         }
75.       },
76.       onerror(err: any) {
77.         errorItemFn(err?.msg || '抱歉，暂无法回答该问题')
78.         console.log('eventSource error: ', `${err}`)
79.         throw err  // 连接遇到http错误时，如跨域等，必须要throw才能停止，不然会一直重连
80.       },
81.       onclose() {
82.         console.log('eventSource close')
83.       },
84.     })
86.   // 终止会话
87.   const stopSession = () => {
88.     abortController.abort()
89.     setAbortController(new AbortController())
90.   }

复制代码

四、常见题目汇总

1、无法添加请求头

应用fetch-event-source办理
2、一个方法需要同时兼容流式和非流式

应用fetch-event-source在onopen阶段处理非流式输出，如报错、接口json返回等
3、碰到跨域时候，请求不停连接

应用fetch-event-source在监听具名事件时，如error，将错误throw err，否则无法中断连接
4、使用SSE时，前端引入markdown渲染时发现后端response data里的前置空格被忽略

服务器响应状态码应该为 200，header 为 Content-Type: text/event-stream，然后保持此连接并以一种特殊的格式写入消息，就像这样：

1. data: Message 1
3. data: Message 2
5. data: Message 3
6. data: of two lines

复制代码

data: 后为消息文本，冒号后面的空格是可选的。
这个可选便是主要缘故原由。查询社区后，找到了两种办理办法：

• 第一种：data后面的用json字符串返回，前端吸收时转成对象拿到里面的content渲染。
  1. ​​​​​​​如：data: {"content":"   1234"}

  复制代码
• 第二种：使用开源社区的@echofly/fetch-event-source插件替换fetch-event-source，微软的这个插件已经好久没维护了，所以对这个题目还没办理。使用方法和fetch-event-source一木一样，只需要更改引入即可。
  参考链接：@echofly/fetch-event-source - npm
  
• 第三种：失败。考虑把空格替换为转义字符，前端无需加任那里理可以直接表现。空格的题目是办理了，但是对代码块等特殊输出带来了影响，使其空格无法展示。
  

具体表现,点击放大查看：

5、fetch方法如何制止

1.   const stopSession = () => {
2.     abortController.abort()
3.     setAbortController(new AbortController())
4.   }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。