告别卡顿！用uni.request的enableChunked实现小程序流式聊天（附完整代码）
告别卡顿！用uni.request的enableChunked实现小程序流式聊天（附完整代码）
在移动应用开发中，流畅的用户体验往往决定了产品的成败。想象一下，当用户在小程序中与AI对话时，如果每次都要等待全部内容加载完成才能显示，那种卡顿感会让最精彩的对话也失去魅力。这正是流式数据传输技术要解决的核心痛点——通过实时逐字输出，让交互变得像真实对话一样自然。
传统的小程序网络请求采用"全量返回"模式，这种"要么全有要么全无"的方式在面对AI生成的长文本、实时数据推送等场景时显得力不从心。而uni-app框架提供的enableChunked功能，配合onChunkReceived事件监听，为我们打开了一扇新的大门。本文将带你深入实战，从原理剖析到代码实现，构建一个完整的流式聊天解决方案。
1. 流式传输的核心原理与技术选型
1.1 HTTP分块传输的本质
HTTP协议中的分块传输编码（Chunked Transfer Encoding）允许服务器将响应分成多个部分逐步发送。这种机制最初是为了解决服务器在发送响应时无法预先确定内容长度的问题，却意外成为了实现实时数据推送的利器。与传统请求相比，它具有三个显著优势：
- 低延迟：首字节到达时间（TTFB）大幅缩短
- 内存友好：不需要缓冲整个响应内容
- 实时性：数据可以边生成边传输
在小程序环境中，由于JavaScript的单线程特性，长时间的数据处理很容易导致界面冻结。流式传输将大数据拆分为小块处理，有效避免了这个问题。
1.2 uni.request的特殊配置
uni-app框架对小程序原生请求API进行了封装，其中几个关键参数决定了流式行为的实现：
| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
enableChunked | Boolean | 否 | 开启分块传输，默认为false |
responseType | String | 否 | 响应数据类型，流式通常设为"text" |
onChunkReceived | Function | 否 | 分块数据接收回调 |
特别需要注意的是，服务器必须支持分块传输编码，常见的实现方式有：
- Node.js的
Transfer-Encoding: chunked响应头 - Python Flask的
stream_with_context - Java的
ChunkedOutput
2. 构建流式请求核心模块
2.1 请求封装与事件监听
创建一个可复用的流式请求模块是项目规范化的第一步。以下是一个经过生产环境验证的实现：
// utils/streamRequest.js let activeRequestTask = null export const createStreamRequest = (config) => { // 取消上一个未完成的请求 if (activeRequestTask) { activeRequestTask.abort() } return new Promise((resolve, reject) => { activeRequestTask = uni.request({ url: config.url, method: config.method || 'GET', data: config.data, header: { 'Accept': 'text/event-stream', 'Content-Type': 'application/json', ...config.headers }, responseType: 'text', enableChunked: true, success: (res) => { resolve(res.data) }, fail: (err) => { reject(err) } }) // 分块数据接收处理 activeRequestTask.onChunkReceived((chunk) => { const data = this._processChunk(chunk) config.onData?.(data) }) }) } // 处理特殊字符和边界情况 function _processChunk(chunk) { let data = chunk.data // 处理可能的base64编码 if (typeof data === 'object' && data.base64) { data = atob(data.base64) } return data.replace(/\0/g, '') // 移除空字符 }2.2 数据拼接与边界处理
流式传输中最棘手的问题是如何正确处理分块边界。以下是几种常见场景的处理策略：
JSON分片拼接：当传输JSON数据时，可能一个完整的JSON对象被拆分成多个块
let buffer = '' function handleJSONChunk(chunk) { buffer += chunk try { const completeData = JSON.parse(buffer) buffer = '' return completeData } catch (e) { // 不完整JSON，等待下次数据 return null } }文本流清洗：处理AI对话中的特殊字符
function cleanTextChunk(text) { return text .replace(/\n+/g, '\n') .replace(/\t+/g, ' ') .replace(/\s+/g, ' ') }二进制流处理：对于非文本数据需要特殊转换
function handleBinaryChunk(arrayBuffer) { const uint8Array = new Uint8Array(arrayBuffer) return String.fromCharCode.apply(null, uint8Array) }
3. 界面实时渲染优化技巧
3.1 高性能的列表更新策略
直接频繁调用setData是小程序性能的大敌。针对流式数据的特点，我们采用增量更新策略：
// 在Page或Component中 let updateTimer = null let pendingUpdates = [] function scheduleUpdate(newText) { pendingUpdates.push(newText) if (!updateTimer) { updateTimer = setTimeout(() => { this.setData({ message: this.data.message + pendingUpdates.join('') }) pendingUpdates = [] updateTimer = null }, 100) // 100ms合并周期 } }这种批处理方式可以将数十次更新合并为一次，性能提升可达300%以上。
3.2 动画与加载状态的配合
流畅的体验离不开视觉反馈。一个专业的实现应该包含：
打字机效果：为逐字输出添加光标动画
.typing-cursor { display: inline-block; width: 2px; height: 1em; background: #333; animation: blink 1s infinite; } @keyframes blink { 0%, 100% { opacity: 1 } 50% { opacity: 0 } }网络状态提示：实时显示传输进度
onChunkReceived(chunk) => { const speed = chunk.size / (chunk.timeStamp - lastChunkTime) this.setData({ loadSpeed: speed.toFixed(2) + 'KB/s' }) }错误恢复机制：自动重试与断点续传
let retryCount = 0 function withRetry(fn, maxRetry = 3) { return async (...args) => { while (retryCount <= maxRetry) { try { return await fn(...args) } catch (err) { if (++retryCount > maxRetry) throw err await new Promise(r => setTimeout(r, 1000 * retryCount)) } } } }
4. 完整实现案例：AI对话组件
4.1 组件化架构设计
将流式聊天封装为独立组件，提高复用性。以下是核心结构：
components/chat/ ├── chat.js # 逻辑层 ├── chat.json # 配置 ├── chat.wxml # 模板 └── chat.wxss # 样式关键实现代码：
// components/chat/chat.js Component({ properties: { apiUrl: String, initialMessage: String }, data: { messages: [], isTyping: false, error: null }, methods: { async sendMessage(content) { this.setData({ isTyping: true }) try { const newMsg = { role: 'user', content } this.data.messages.push(newMsg) await createStreamRequest({ url: this.properties.apiUrl, method: 'POST', data: { messages: this.data.messages }, onData: (chunk) => { this._appendToLastMessage(chunk) } }) } catch (err) { this.setData({ error: err.message }) } finally { this.setData({ isTyping: false }) } }, _appendToLastMessage(text) { const lastIdx = this.data.messages.length - 1 const key = `messages[${lastIdx}].content` this.setData({ [key]: (this.data.messages[lastIdx].content || '') + text }) } } })4.2 服务端适配方案
不同的后端框架需要特定的配置才能支持分块传输。以下是常见技术的实现片段：
Node.js (Express):
app.get('/stream', (req, res) => { res.setHeader('Content-Type', 'text/event-stream') res.setHeader('Transfer-Encoding', 'chunked') const interval = setInterval(() => { res.write(`data: ${JSON.stringify({ chunk: Date.now() })}\n\n`) }, 1000) req.on('close', () => clearInterval(interval)) })Python (Flask):
@app.route('/stream') def stream(): def generate(): for i in range(10): yield f"data: Message {i}\n\n" time.sleep(1) return Response(generate(), mimetype='text/event-stream')Java (Spring Boot):
@GetMapping("/stream") public SseEmitter streamData() { SseEmitter emitter = new SseEmitter(); ExecutorService executor = Executors.newSingleThreadExecutor(); executor.execute(() -> { try { for (int i = 0; i < 10; i++) { emitter.send(SseEmitter.event() .data("Chunk " + i) .id(String.valueOf(i))); Thread.sleep(1000); } emitter.complete(); } catch (Exception ex) { emitter.completeWithError(ex); } }); return emitter; }5. 性能调优与异常处理
5.1 内存泄漏预防
长时间运行的流式连接容易积累内存问题。关键防护措施包括：
请求生命周期管理：
let requestTasks = new Set() function addRequestTask(task) { requestTasks.add(task) task.onComplete = () => requestTasks.delete(task) } function abortAllRequests() { requestTasks.forEach(task => task.abort()) requestTasks.clear() } // 在页面onUnload中调用 abortAllRequests()数据量监控：
const MAX_BUFFER_SIZE = 1024 * 1024 // 1MB let bufferSize = 0 function checkBuffer(chunk) { bufferSize += chunk.length if (bufferSize > MAX_BUFFER_SIZE) { throw new Error('Buffer overflow') } }
5.2 网络异常处理策略
移动网络环境复杂，需要完善的错误恢复机制：
自动重试策略：
async function resilientRequest(config, retries = 3) { for (let i = 0; i < retries; i++) { try { return await createStreamRequest(config) } catch (err) { if (i === retries - 1) throw err await new Promise(r => setTimeout(r, 1000 * (i + 1))) } } }心跳检测：
let lastChunkTime = Date.now() function startHeartbeatCheck() { const interval = setInterval(() => { if (Date.now() - lastChunkTime > 30000) { // 30秒无数据 reconnect() } }, 5000) return () => clearInterval(interval) }离线缓存：
function savePartialResponse(partialData) { try { uni.setStorageSync('stream_cache', { timestamp: Date.now(), data: partialData }) } catch (e) { console.error('Storage failed', e) } }
6. 高级应用场景扩展
6.1 大文件分块上传
同样的技术原理可以逆向应用于文件上传场景：
function uploadInChunks(filePath, chunkSize = 1024 * 512) { const fileTask = uni.uploadFile({ url: 'https://api.example.com/upload', filePath, name: 'file', formData: { chunkSize, totalSize: fileInfo.size }, enableChunked: true, onChunkProgress: (res) => { console.log(`Uploaded ${res.loaded}/${res.total}`) } }) return fileTask }6.2 实时音视频元数据传输
结合WebSocket与分块传输，实现低延迟的媒体流控制：
function setupMediaStream() { const socket = new WebSocket('wss://stream.example.com') const mediaChunks = [] socket.onmessage = (event) => { const chunk = decodeMediaChunk(event.data) mediaChunks.push(chunk) if (mediaChunks.length > 5) { // 5个chunk组成一帧 renderMediaFrame(mediaChunks.splice(0, 5)) } } }6.3 服务端推送事件(SSE)集成
对于需要长时间连接的场景，SSE是更专业的选择：
function setupSSEConnection() { const eventSource = new EventSource('/sse-endpoint') eventSource.addEventListener('message', (e) => { updateUI(JSON.parse(e.data)) }) eventSource.addEventListener('error', () => { setTimeout(setupSSEConnection, 5000) // 5秒后重连 }) }在实现这些高级场景时，关键是要理解流式传输的核心优势——实时性和资源利用率，根据具体业务需求选择最适合的技术组合。比如对于金融实时行情，可能需要WebSocket+分块传输的双重保障；而对于日志推送，单纯的SSE可能就足够了。