前言

最近需要使用uniapp开发一个智能对话页面，其中就需要使用SSE进行通信。

本文介绍下在uniapp中如何基于uni.request实现SSE流式处理。

在线体验

#小程序:yinuosnowball

SSE传输格式

返回输出的流式块:

Content-Type为text/event-stream
每个流式块均为 data: 开头，块之间以 \n\n 即两个换行符分隔, 如下所示：
后端接口定义的数据如下： - event为message，开始接收数据，answer为返回的结果 - event为message_end结束

在这里插入图片描述

接口数据已约定完成，下一步使用uniapp开始接收处理数据。

uniapp处理数据

客户端实现在微信小程序中接收 SSE 流式响应，需要以下步骤：

配置 HTTP 请求：设置适当的请求头和参数，以确保服务器返回流式响应。
处理分块数据：由于SSE是分块传输的，我们需要监听每个数据块，并解析它们。
错误处理：当每一次返回的最后出现不是完整的响应时，需要进行特殊处理。
完成时：可以进行追问等额外处理

下面使用 uni.request实现SSE的案例：

基本框架：

const requestTask = uni.request({url,method: 'POST',header: {Accept: 'text/event-stream',Authorization,},data,enableChunked: true,responseType: 'arraybuffer',success: (res) => {console.log('Data received 数据接受完毕:', res.data)},fail: (error) => {console.log('打印***error 错误处理', error)},complete: (complete) => {console.log('打印***complete 完成接收', complete)}
})requestTask.onChunkReceived((res)=>{// 处理数据
})

通过对requestTask的onChunkReceived监听就可以得到数据块，通过打印我们可以看到数据返回是ArrayBuffer，我们需要进行处理。

在这里插入图片描述

 const uint8Array = new Uint8Array(res.data);let text = String.fromCharCode.apply(null, uint8Array);

解析后得到以data:data:返回的格式，

此处需要注意解析后是data: 还是 data:data:格式

在这里插入图片描述

进一步处理：由于返回的数据块不是一段一段，而是很多段都返回，因此我们需要进行\n\n进行拆分，然后逐个解析：

在这里插入图片描述

const arr = text.split('\n\n').filter(Boolean)
arr.forEach(msg => {const jsonStr = msg.substring(11); // 去掉 'data:data: ' 前缀const data = JSON.parse(jsonStr);switch (data.event) {case 'message': {// 拼接返回文本this.dialogueList[existingMessageIndex].answer += data.answer;break;}case 'message_end':// 消息结束break;}
});

至此，我们就可以接收到消息，如果就这样那就最好，但对接的过程发现，每一次返回的文本最后一段不是完整的，导致解析出现失败，如下图所示：

在这里插入图片描述

解决方案是：定义一个变量，当解析出现失败时，肯定是最后一段，进行存储，下一次接收到数据将上一次存储的进行拼接，然后解析:

在这里插入图片描述

具体代码：

// 每次发送存储数据
const msgObj = {query,answer: "",conversationId: null,isDone: false
}
this.dialogueList.push(msgObj)
// 保存上一次失败的text
let lastText = ''
requestTask.onChunkReceived((res) => {// 第一步：获取 字符串 数组const uint8Array = new Uint8Array(res.data);let text = lastText + String.fromCharCode.apply(null, uint8Array);lastText = '';let arr = text.split('\n\n').filter(Boolean)let lastIndex = arr.length - 1// 第二步：是否可以直接进行解析try {// 判断是否可以全部解析完成arr.every(item => JSON.parse(item.substring(11)))} catch (error) {// 如果报错截取最后一项lastText = arr[lastIndex]arr = arr.filter((_, i) => i !== lastIndex)}// 处理数据块if (arr.length) {try {arr.forEach(msg => {const jsonStr = msg.substring(11); // 去掉 'data: ' 前缀const data = JSON.parse(jsonStr);const existingMessageIndex = this.dialogueList.findIndex(item => item === msgObj);switch (data.event) {case 'message': {// 查找是否存在相同ID的消息this.dialogueList[existingMessageIndex].answer += data.answer;break;}case 'message_end':// 消息结束break;}});} catch (error) {console.error('解析数据失败:', error);}}
});

这样就完成了对不连续返回的错误处理。

如果需要直接结束请求，可以直接使用requestTask.abort()。

完整代码

function sendMsg(query) {const msgObj = {query, // 问题answer: "", // 回答的结果conversationId: null,feedback: null,isDone: false // 自定义格式，用于加载处理}this.dialogueList.push(msgObj)// 请求参数const data = {}this.requestTask = uni.request({url,method: 'POST',header: {Accept: 'text/event-stream',Authorization: getStorage(tokenKeyEnum.zhonglv),},data,enableChunked: true,responseType: 'arraybuffer',success: (res) => {console.log('Data received 数据接受完毕:', res.data)},fail: (error) => {console.log('打印***error 错误处理', error)},complete: (complete) => { console.log('打印***complete 完成接收', complete)}})let lastText = ''this.requestTask.onChunkReceived((res) => {// 第一步：获取 字符串 数组const uint8Array = new Uint8Array(res.data);let text = lastText + String.fromCharCode.apply(null, uint8Array);lastText = '';let arr = text.split('\n\n').filter(Boolean)let lastIndex = arr.length - 1// 第二步：是否可以直接进行解析try {let isCanResolve = arr.every(item => JSON.parse(item.substring(11)))console.log('打印***isCanResolve', isCanResolve)} catch (error) {// 如果报错截取最后一项lastText = arr[lastIndex]arr = arr.filter((_, i) => i !== lastIndex)console.log('打印***error', error)}// 处理数据块if (arr.length) {try {arr.forEach(msg => {const jsonStr = msg.substring(11); // 去掉 'data:data: ' 前缀const data = JSON.parse(jsonStr);const existingMessageIndex = this.dialogueList.findIndex(item => item === msgObj);switch (data.event) {case 'message': {// 查找是否存在相同ID的消息this.dialogueList[existingMessageIndex].answer += data.answer;break;}case 'message_end':// 消息结束this.dialogueList[existingMessageIndex].isDone = true;break;}});} catch (error) {console.error('解析数据失败:', error);}}});
}