使用 UniApp 在微信小程序中实现 SSE 流式响应

在这里插入图片描述

概述

服务端发送事件（Server-Sent Events, SSE）是一种允许服务器向客户端推送实时更新的技术。SSE 提供了一种单向的通信通道，服务器可以持续地向客户端发送数据，而不需要客户端频繁发起请求。这对于需要实时更新的应用场景非常有用。

流式传输的特点是将数据逐步传输给客户端，而不需要等待完整的响应生成。这意味着在传输过程中，数据会逐步发送给客户端，而不是一次性发送所有数据，对于基于文本的AI对话来说，这意味着每个单词或短语可以随着模型预测它们时即时显示出来，从而营造出一种更加自然和动态的交流体验。

最近在对接大模型对话生成接口，查找官方文档中并没有找到明确的实现说明，本文根据 Uniapp 及微信小程序开发文档，基于 uni.request 实现了一个简单的SSE客户端。

准备

服务端已提供SSE接口，可通过 Apifox 直接访问进行测试。

Uniapp 客户端实现

要实现在微信小程序中接收 SSE 流式响应，我们需要做几个关键步骤：

配置 HTTP 请求：设置适当的请求头和参数，以确保服务器知道我们期望的是流式响应。
处理分块数据：由于 SSE 是分块传输的，我们需要监听每个数据块，并适当地解析它们。
错误和完成处理：定义当遇到错误或完成时的行为。

下面是一个使用 uni.request API 实现 SSE 的例子：


let buffer = ''
function decode(data: ArrayBuffer): string {// 根据协议对数据进行解析，省略...// 注意数据可能是不连续的，需要通过 buffer 进行拼接
}function streamPost(url, data, onData, onError = null, onComplete = null) {function onChunkReceived(res) {onData(decode(res.data))}function onHeadersReceived(res) {console.log('onHeadersReceived', res)}const requestTask = uni.request({url: baseUrl + apiPath + url,method: 'POST',header: {Accept: 'text/event-stream', // 确保服务器知道我们期望的是流式响应Authorization: uni.getStorageSync('token'),// ...其他参数},data,enableChunked: true, // onChunkReceived, 否则走success()responseType: 'arraybuffer',success: (res) => {console.log('Data received:', res.data) // 开启 enableChunked 时仅最后一次会走这个},fail: (error) => { // 错误处理if (onError) {onError(error)}console.error('SSE failed:', error)},complete: () => { // 完成接收if (onComplete) {onComplete()}if (onHeadersReceived) {requestTask?.offHeadersReceived(onHeadersReceived)}if (onChunkReceived) {// @ts-expect-error uni-app types lostrequestTask?.offChunkReceived(onChunkReceived)}},})if (onHeadersReceived) {requestTask.onHeadersReceived(onHeadersReceived)}if (onChunkReceived) {// @ts-expect-error uni-app types lostrequestTask.onChunkReceived(onChunkReceived) // 注册数据接收响应函数}return requestTask // 外部可通过 requestTask.abort(); 主动结束
}

在 nginx 中开启transfer_encoding, 同时关闭缓存 proxy_buffering。

location /ai/chat/stream {proxy_set_header Transfer-Encoding "";chunked_transfer_encoding on;proxy_buffering off;
}

总结

开启：enableChunked: true
设置请求 Header：Accept: 'text/event-stream'
注册数据接收响应函数: requestTask.onChunkReceived(onChunkReceived)
主动结束： requestTask.abort()
分块数据解析：decode()

通过以上步骤，我们成功地在 UniApp 中实现了 SSE 流式响应，增强了应用程序的实时交互能力。希望这篇文章能为你在 UniApp 中集成实时数据更新功能提供有价值的参考。

参考

uniapp api 文档: https://uniapp.dcloud.net.cn/api/request/request.html
小程序开发文档：wx.request https://developers.weixin.qq.com/miniprogram/dev/api/network/request/RequestTask.onChunkReceived.html