前端-前端uniapp在微信小程序中实现 SSE 流式响应

前言

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

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

SSE传输格式

返回输出的流式块:

Content-Type为text/event-stream
每个流式块均为 data: 开头，块之间以 \n\n 即两个换行符分隔, 如下所示：

3. 后端接口定义的数据如下： - 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);
      }
    }
  });
}