微信小程序对接SSE接口的实现方案

文
章
目
录

在微信小程序开发中，经常会遇到需要实时接收服务端推送数据的场景。然而，微信小程序原生API并不支持标准的Server-Sent Events（SSE）。为了解决这个问题，本文将介绍一种基于wx.request结合enableChunked特性实现的自定义SSE流数据接收方案。

一、方案核心功能与特点

（一）流式数据处理

利用wx.request的onChunkReceived功能，能够监听分块数据，从而实现实时接收服务端推送的流数据。在数据接收过程中，还能将接收到的ArrayBuffer格式数据转换为字符串，并有效处理不完整字符问题，确保数据的完整性。

（二）不完整字符处理

为了处理不完整字符，方案中使用lastArrayBuffer来缓存上一次未解析完整的字节数据。每次新数据到来时，将其与缓存数据合并后再进行解析。特别是针对UTF-8多字节字符可能被截断的情况，会检测并保存剩余字节，等待后续数据进行拼接，以此避免出现乱码或解析失败的问题。

（三）SSE数据解析

通过特定的正则表达式（/data:([\s\S]*?)(?=\n\s*data:|$)/g），可以精准提取以data:开头的消息块。在解析JSON字符串时，使用safeJsonParse方法，该方法能够安全地解析JSON字符串，并在遇到异常情况时提供错误提示。

（四）遗留消息管理

引入LegacyMessage机制，用于缓存解析失败的半条消息。这些消息会与后续接收的数据进行拼接，然后重新解析，确保消息不会丢失。在处理多块数据时，会逐块解析，并对最后一块数据进行特殊处理，判断其是否完整。

（五）接口封装与使用

提供了SSE_WX函数，在使用时可配置url、data、success、error和finish等回调。该函数返回abort方法，用于手动中断请求并清理监听，大大增强了使用的灵活性。通过随机生成的uid来区分不同请求，有效避免数据混淆。

（六）异常处理与清理

在请求完成或中断时，会自动清理LegacyMessage和lastArrayBuffer缓存。同时，能够捕获并记录请求过程中出现的异常，方便后续调试和优化。

二、使用场景

此方案适用于各种需要实时接收服务端推送数据的微信小程序场景，比如聊天消息的实时获取、实时通知的接收，以及数据流更新等情况。它有效解决了微信小程序对SSE支持不足的问题，同时保证了数据解析的可靠性和完整性。

三、引用示例

下面是一个使用该方案的简单示例：

SSE_WX({
  url: '/api/sse',
  data: { id: 123 },
  success: (data) => console.log('收到数据:', data),
  error: (err) => console.error('错误:', err),
  finish: (res) => console.log('完成:', res)
}).abort() // 可随时中断

在上述示例中，通过SSE_WX函数发起请求，并配置了相应的回调函数，用于处理接收数据、错误和请求完成的情况。还可以随时调用abort方法中断请求。

四、完整封装代码解析

import { BASE_URL } from '../config'

// 存储上次解析时剩余的不完整数据
let lastArrayBuffer = new Map()

// arrayBuffer 转 String，处理不完整字符的情况
function arrayBufferToString(arr, uid) {
  // 如果已经是字符串，直接返回
  if (typeof arr === 'string') {
    return arr
  }

  // 合并上次的不完整数据（如果有）
  if (lastArrayBuffer.has(uid)) {
    console.log('存在未解析字段')

    // 创建一个新的 ArrayBuffer 来合并数据
    const combinedLength = lastArrayBuffer.get(uid).byteLength + arr.byteLength
    const combinedBuffer = new ArrayBuffer(combinedLength)
    const combinedView = new Uint8Array(combinedBuffer)

    // 复制上次剩余的数据
    combinedView.set(new Uint8Array(lastArrayBuffer.get(uid)), 0)
    // 复制新的数据
    combinedView.set(new Uint8Array(arr), lastArrayBuffer.get(uid).byteLength)

    // 使用合并后的数据替换当前输入
    arr = combinedBuffer
    // 清空 lastArrayBuffer，因为已合并
    lastArrayBuffer.delete(uid)
  }

  const dataview = new DataView(arr)
  const ints = new Uint8Array(arr.byteLength)
  for (let i = 0; i < ints.length; i++) {
    ints[i] = dataview.getUint8(i)
  }

  let str = ''
  let _arr = ints
  let i = 0

  while (i < _arr.length) {
    if (_arr[i]) {
      let one = _arr[i].toString(2).padStart(8, '0')
      let v = one.match(/^1+?(?=0)/)

      if (v && one.length == 8) {
        const bytesLength = v[0].length

        // 检查是否有足够的字节来完成这个多字节字符
        if (i + bytesLength - 1 >= _arr.length) {
          // 不完整字符，保存剩余数据到 lastArrayBuffer
          const remainingBytes = _arr.slice(i)
          const buffer = new ArrayBuffer(remainingBytes.length)
          new Uint8Array(buffer).set(remainingBytes)
          lastArrayBuffer.set(uid, buffer)
          console.log('==================出现半个字符解析==================')
          break // 中断处理
        }

        let store = _arr[i].toString(2).slice(7 - bytesLength)
        let validSequence = true

        for (let st = 1; st < bytesLength; st++) {
          const continuationByteIndex = st + i
          if (continuationByteIndex < _arr.length) {
            // 验证后续字节是否是有效的 UTF-8 延续字节（以 10 开头）
            const nextByte = _arr[continuationByteIndex].toString(2).padStart(8, '0')
            if (nextByte.slice(0, 2) === '10') {
              store += nextByte.slice(2)
            } else {
              validSequence = false
              break
            }
          } else {
            validSequence = false
            break
          }
        }

        if (validSequence) {
          str += String.fromCharCode(parseInt(store, 2))
          i += bytesLength
        } else {
          // 无效的 UTF-8 序列，当作单字节处理
          str += String.fromCharCode(_arr[i])
          i++
        }
      } else {
        // 单字节字符
        str += String.fromCharCode(_arr[i])
        i++
      }
    } else {
      // 0 值字节
      i++
    }
  }

  return str
}

function parseSSEData(sseData) {
  // 使用正则表达式匹配每个data:开头的块，包括可能的多行内容
  const regex = /data:([\s\S]*?)(?=\n\s*data:|$)/g
  const matches = [...sseData.matchAll(regex)]

  // 从匹配结果中提取JSON字符串
  const jsonStrings = matches.map(match => {
    // 获取匹配的内容并清理
    const jsonContent = match[0].trim().replace(/\n/g, '')
    return jsonContent
  })

  return jsonStrings
}

// 解析`data:`开头的json字符串
const safeJsonParse = (str = '') => {
  const str1 = str.trim()
  if (str1.startsWith('data:')) {
    try {
      const data = JSON.parse(str1.slice(5))
      return data
    } catch (err) {
      throw new Error('[json解析失败]')
    }
  } else {
    throw new Error('[未匹配到消息头]')
  }
}

/**
 * 遗留消息
 * 如果一条消息解析失败，则认为该消息为半条消息，和后续消息拼接后再进行解析
 */
const LegacyMessage = new Map()

// 微信小程序实现sse,通过wx自己的方式实现 -- 该接口有一个明显的问题，同时只能触发一次
export const SSE_WX = ({ url, data, success, error, finish }) => {
  // lastArrayBuffer = null // 重置半个流数据
  // 接口赋值
  let requestTask = null
  try {
    const uid = Math.random().toString(36).substring(2, 9)

    // 处理接收到的数据
    const listener = res => {
      // 1. 转换成字符串的格式
      const str1 = arrayBufferToString(res.data, uid)
      console.log('------------------------------------------')
      console.log('接收消息:\n', str1)
      // 明确区分空字符串和 null
      if (!str1 && str1!== '') {
        return
      }
      // 2. 判断是否存在未解析部分，如果存在，则解析合并后的字符串
      let prefix = ''
      if (LegacyMessage.has(uid)) {
        prefix = LegacyMessage.get(uid)
      }
      const str2 = `${prefix}${str1}`
      // 3. 进行解析
      const jsonStrings = parseSSEData(str2)
      if (!jsonStrings.length) {
        // 3.1. 如果解析为空 则代表该部分为片段部分
        LegacyMessage.set(uid, str2)
      } else {
        // 3.2. 解析内容不为空
        LegacyMessage.delete(uid)
        // 4.1. 判断解析后数组，是否是完整的数据，最后一项进行特殊处理
        for (let i = 0; i < jsonStrings.length - 1; i++) {
          const data = safeJsonParse(jsonStrings[i])
          success && success(data)
        }
        // 4.2. 最后一项特殊处理，判断正常解析，还是记录未处理的内容
        const last = jsonStrings[jsonStrings.length - 1]
        try {
          const data = safeJsonParse(last)
          success && success(data)
        } catch (err) {
          LegacyMessage.set(uid, last)
        }
      }
    }

    // 发起请求
    requestTask = wx.request({
      url: `${BASE_URL}${url}`,
      method: 'POST',
      enableChunked: true, // enableChunked必须为true
      data: data,
      header: {
        'content-type': 'application/json'
      },
      // 执行完成
      complete(res) {
        LegacyMessage.delete(uid)
        lastArrayBuffer.delete(uid)
        // 触发完成回调
        if (finish && typeof finish === 'function') {
          finish(res)
        }
      }
    })
    // 监听服务端返回的数据
    requestTask.onChunkReceived(listener)

    return {
      abort: () => {
        // 移除监听 需传入与监听时同一个的函数对象
        requestTask.offChunkReceived(listener)
        requestTask.abort()
      }
    }
  } catch (err) {
    console.error('[sse请求异常]', err)
    error(err)
    requestTask.abort()
  }
}

数据转换函数arrayBufferToString：该函数负责将ArrayBuffer格式的数据转换为字符串。如果存在上次未解析完的数据（通过lastArrayBuffer判断），会先将其与新数据合并。在转换过程中，会处理UTF-8多字节字符，遇到不完整字符时，将剩余字节保存到lastArrayBuffer 。
SSE数据解析函数parseSSEData：利用正则表达式，从接收到的SSE数据中提取出以data:开头的消息块，并将其整理成JSON字符串数组。
JSON安全解析函数safeJsonParse：用于解析以data:开头的JSON字符串。如果解析过程中出现异常，会抛出相应的错误提示。
LegacyMessage机制：使用Map数据结构来存储解析失败的半条消息。当新数据到来时，会将其与缓存的消息拼接后再次解析。
SSE_WX函数：这是对外提供的核心函数，用于发起SSE请求。在函数内部，首先生成一个随机的uid ，用于区分不同请求。通过wx.request发起请求，并设置enableChunked: true以支持分块接收数据。在onChunkReceived回调中处理接收到的数据，根据解析结果调用相应的回调函数。同时返回abort方法，用于中断请求和清理监听。

通过以上方案，微信小程序开发者可以在不依赖原生SSE支持的情况下，实现高效可靠的流数据接收功能，满足各种实时数据交互的需求。

版权声明：本站文章，如无说明，均为本站原创，转载请注明文章来源。如有侵权，请联系博主删除。
本文链接：https://www.panziye.com/front/17389.html