uni-app 流式输出实战：enableChunked + SSE 打通小程序与 H5 大模型对话

前段时间把一个 AI 陪聊形态的小程序从「转圈等 5 秒、然后一次性吐出一大坨字」改成了逐字往外冒。听起来只是换个请求方式，实际上从抓包到真机验证折腾了两天。中间最耗时的不是接 API，而是三件很小但很恶心的事：多字节汉字被切成两半、SSE 事件被拆到两个 chunk 里、以及每收到一个分片就改一次响应式数据导致列表抖成筛子。

这篇把整套管线拆开写一遍，代码可以直接抄走用。

问题出在哪：小程序没有 EventSource

浏览器里做流式输出很轻松，EventSource 或者 fetch 拿到 ReadableStream 就完事。但这两个东西小程序运行时里都没有。小程序也不是完全不给活路，它在基础库 2.20.2 之后给 RequestTask 加了一个 onChunkReceived 回调，配合请求参数 enableChunked: true，服务端每 flush 一次数据，你就能拿到一个 ArrayBuffer。

这就是全部家当了。没有 TextDecoder（老版本没有，新版本有但兼容性不齐），没有内置的 SSE 解析，拿到手的是一堆裸字节。

所以整条链路是这样的：

uni.request(enableChunked)
        ↓
   onChunkReceived(ArrayBuffer)
        ↓
   ① UTF-8 增量解码（处理跨包多字节字符）
        ↓
   ② SSE 帧切分（处理跨包事件分隔符）
        ↓
   ③ 单帧字段解析（data / event / [DONE]）
        ↓
   ④ JSON 提取 delta.content
        ↓
   ⑤ 渲染节流（40ms 批量刷一次）
        ↓
      视图

下面一步一步来。

第一步：把 uni.request 变成一条不关闭的水管

关键参数就两个：enableChunked: true 和 responseType: 'arraybuffer'。前者打开分片能力，后者保证 success 回调里拿到的也是二进制而不是被提前转成字符串的残体。

const task = uni.request({
  url: 'https://your-api.com/v1/chat/completions',
  method: 'POST',
  enableChunked: true,
  responseType: 'arraybuffer',
  header: {
    'Content-Type': 'application/json',
    'Accept': 'text/event-stream'
  },
  data: {
    model: 'gpt-4o-mini',
    messages: [{ role: 'user', content: '你好' }],
    stream: true
  },
  success: () => {
    // 注意：这里的 res.data 是整包聚合结果
    // 我们已经通过 onChunkReceived 增量消费过了，不再处理
  },
  fail: (err) => {
    console.error('请求失败', err)
  }
})

if (typeof task.onChunkReceived !== 'function') {
  // 老基础库 / 部分小程序平台没有这个能力，走降级
  task.abort()
  throw new Error('当前环境不支持分块返回')
}

task.onChunkReceived((res) => {
  // res.data 是 ArrayBuffer，不是字符串
  console.log(res.data.byteLength)
})

有两个点容易忽略。一是 success 回调照样会触发，而且触发在最后，别在里面重复追加内容，否则整段回答会被写两遍。二是 enableChunked 目前主要在微信小程序端和 App 端可用，支付宝小程序、部分快应用上没有对应实现，所以能力探测这一步不能省。

第二步：UTF-8 增量解码，最容易翻车的地方

拿到 ArrayBuffer 之后第一反应通常是转成字符串。小程序的 TextDecoder 支持情况不稳定，很多人会写一个手动的转换：

// 这种写法在流式场景里一定会出问题
function ab2str(buffer) {
  return String.fromCharCode.apply(null, new Uint8Array(buffer))
}

问题在于，服务端 flush 的边界是按缓冲区大小来的，不是按字符边界来的。一个 UTF-8 汉字占 3 个字节，em 表情占 4 个字节，它们完全可能被劈成 [0xE4, 0xBD] 和 [0xA0] 两片分两次发过来。上面那段代码会把第一个分片解成一个乱码字符，第二个分片再解成一个乱码字符——屏幕上就会莫名多出几个「锟斤拷」。

正确做法是带上「残留字节」的状态机：遇到不完整的序列就把尾部字节存起来，等下一次分片拼上再解。

function createUTF8Decoder() {
  // 上一次没解完的尾部字节
  let carry = new Uint8Array(0)

  return {
    decode(arrayBuffer) {
      const incoming = new Uint8Array(arrayBuffer)

      // 把上次残留的字节拼到最前面
      let bytes
      if (carry.length) {
        bytes = new Uint8Array(carry.length + incoming.length)
        bytes.set(carry, 0)
        bytes.set(incoming, carry.length)
      } else {
        bytes = incoming
      }

      let out = ''
      let i = 0

      while (i < bytes.length) {
        const b = bytes[i]

        // ASCII 直接过
        if (b < 0x80) {
          out += String.fromCharCode(b)
          i += 1
          continue
        }

        // 判断这个首字节后面还跟着几个字节
        let extra = 0
        let cp = 0
        if ((b & 0xe0) === 0xc0) {        // 110xxxxx → 2 字节
          extra = 1
          cp = b & 0x1f
        } else if ((b & 0xf0) === 0xe0) { // 1110xxxx → 3 字节
          extra = 2
          cp = b & 0x0f
        } else if ((b & 0xf8) === 0xf0) { // 11110xxx → 4 字节
          extra = 3
          cp = b & 0x07
        } else {
          // 非法首字节，跳过，避免死循环
          i += 1
          continue
        }

        // 剩余字节不够，说明这个字符被切开了，留给下一次
        if (i + extra >= bytes.length) break

        let valid = true
        for (let k = 1; k <= extra; k++) {
          const nb = bytes[i + k]
          if ((nb & 0xc0) !== 0x80) {
            valid = false
            break
          }
          cp = (cp << 6) | (nb & 0x3f)
        }

        if (!valid) {
          // 不是合法的续字节，丢弃当前首字节继续
          i += 1
          continue
        }

        out += String.fromCodePoint(cp)
        i += extra + 1
      }

      // 把没消费完的尾部存起来，最多 3 个字节
      const rest = bytes.slice(i)
      carry = rest.length > 3 ? new Uint8Array(0) : rest

      return out
    },

    reset() {
      carry = new Uint8Array(0)
    }
  }
}

这段代码里 if (i + extra >= bytes.length) break 是整个函数的核心。它表示「这个字符的后半截还没到」，此时直接跳出循环，把从 i 开始的所有字节塞进 carry，下一次再拼上。

另外 carry 做了上限保护。正常情况下它最多存 3 个字节，如果超过说明流里出现了非法数据，直接清空避免无限堆积。

第三步：SSE 帧的切分与解析

解码出文本之后，不能直接 split('nn')。原因和上面一样——分隔符本身可能被拆到两个分片里，你这一片结尾只有 n，下一片开头才是另一个 n。

所以同样要用缓冲区：不够切就先攒着。

function createSSEParser() {
  let buffer = ''

  return {
    push(text) {
      buffer += text
      const frames = []

      while (true) {
        const a = buffer.indexOf('nn')
        const b = buffer.indexOf('rnrn')

        let idx = -1
        let sepLen = 2

        if (a !== -1 && b !== -1) {
          if (b < a) { idx = b; sepLen = 4 } else { idx = a; sepLen = 2 }
        } else if (a !== -1) {
          idx = a
          sepLen = 2
        } else if (b !== -1) {
          idx = b
          sepLen = 4
        } else {
          break
        }

        frames.push(buffer.slice(0, idx))
        buffer = buffer.slice(idx + sepLen)
      }

      return frames
    },

    reset() {
      buffer = ''
    }
  }
}

拿到一个完整帧之后，按 SSE 规范解析字段。data: 后面的内容才是载荷，冒号后如果有一个空格要去掉；以冒号开头的行是注释，很多服务端用它当心跳，直接忽略。

function parseFrame(frame) {
  const lines = frame.split(/r?n/)
  const dataLines = []
  let done = false

  for (let i = 0; i < lines.length; i++) {
    const line = lines[i]

    // 空行、心跳注释行跳过
    if (!line || line.charAt(0) === ':') continue

    const colon = line.indexOf(':')
    const field = colon === -1 ? line : line.slice(0, colon)
    let value = colon === -1 ? '' : line.slice(colon + 1)

    // SSE 规范：冒号后紧跟一个可选空格
    if (value.charAt(0) === ' ') value = value.slice(1)

    if (field === 'data') {
      if (value === '[DONE]') {
        done = true
        continue
      }
      dataLines.push(value)
    }
  }

  if (done) return { done: true, payload: null }
  if (!dataLines.length) return { done: false, payload: null }

  return { done: false, payload: dataLines.join('n') }
}

注意 dataLines.join('n') 这一句。SSE 允许一个事件里带多行 data:，规范要求用换行拼接。虽然 OpenAI 兼容格式的服务端基本不会这么发，但加上这行几乎没成本，遇上时也不至于解析失败。

第四步：把整条链路封装成 composable

上面三块是纯函数，可以单独放一个文件。真正组合起来的部分用一个 composable 收口，页面里只关心 content 和 loading。

// composables/useChatStream.js
import { ref } from 'vue'

export function useChatStream(options) {
  const {
    url,
    headers = {},
    model = '',
    extraBody = {}
  } = options || {}

  const content = ref('')
  const loading = ref(false)
  const errorMsg = ref('')

  let requestTask = null
  let abortController = null
  let decoder = null
  let parser = null
  let active = false

  // 渲染节流用的待刷缓冲区
  let pending = ''
  let flushTimer = null

  function scheduleFlush() {
    if (flushTimer) return
    flushTimer = setTimeout(() => {
      flushTimer = null
      if (!pending) return
      content.value += pending
      pending = ''
    }, 40)
  }

  function flushNow() {
    if (flushTimer) {
      clearTimeout(flushTimer)
      flushTimer = null
    }
    if (pending) {
      content.value += pending
      pending = ''
    }
  }

  function handleText(text) {
    const frames = parser.push(text)

    for (let i = 0; i < frames.length; i++) {
      const { done, payload } = parseFrame(frames[i])

      if (done) {
        finish()
        return
      }

      if (!payload) continue

      let json
      try {
        json = JSON.parse(payload)
      } catch (e) {
        // 极少数服务端会发非 JSON 的 data 行，丢掉即可
        continue
      }

      const choice = json.choices && json.choices[0]
      const delta = choice && choice.delta && choice.delta.content

      if (delta) {
        pending += delta
        scheduleFlush()
      }
    }
  }

  function finish() {
    if (!active) return
    active = false
    flushNow()
    loading.value = false
    requestTask = null
    abortController = null
  }

  function stop() {
    active = false
    flushNow()
    if (requestTask && typeof requestTask.abort === 'function') {
      try { requestTask.abort() } catch (e) {}
    }
    if (abortController) {
      try { abortController.abort() } catch (e) {}
    }
    requestTask = null
    abortController = null
    loading.value = false
  }

  function send(messages) {
    stop()

    content.value = ''
    errorMsg.value = ''
    pending = ''
    loading.value = true
    active = true

    decoder = createUTF8Decoder()
    parser = createSSEParser()

    const body = Object.assign({
      model,
      messages,
      stream: true
    }, extraBody)

    // #ifdef H5
    return sendByFetch(body)
    // #endif

    // #ifndef H5
    return sendByChunked(body)
    // #endif
  }

  function sendByChunked(body) {
    return new Promise((resolve) => {
      let settled = false
      const done = () => {
        if (settled) return
        settled = true
        finish()
        resolve()
      }

      requestTask = uni.request({
        url,
        method: 'POST',
        enableChunked: true,
        responseType: 'arraybuffer',
        header: Object.assign({
          'Content-Type': 'application/json',
          'Accept': 'text/event-stream'
        }, headers),
        data: body,
        success: () => done(),
        fail: (err) => {
          errorMsg.value = (err && err.errMsg) || '请求失败'
          done()
        }
      })

      if (!requestTask || typeof requestTask.onChunkReceived !== 'function') {
        errorMsg.value = '当前运行环境不支持分块返回，请升级基础库'
        if (requestTask && requestTask.abort) requestTask.abort()
        done()
        return
      }

      requestTask.onChunkReceived((res) => {
        if (!active) return
        const text = decoder.decode(res.data)
        if (text) handleText(text)
      })
    })
  }

  async function sendByFetch(body) {
    abortController = new AbortController()

    try {
      const resp = await fetch(url, {
        method: 'POST',
        headers: Object.assign({
          'Content-Type': 'application/json',
          'Accept': 'text/event-stream'
        }, headers),
        body: JSON.stringify(body),
        signal: abortController.signal
      })

      if (!resp.ok) throw new Error('HTTP ' + resp.status)

      const reader = resp.body.getReader()
      const textDecoder = new TextDecoder('utf-8')

      while (active) {
        const { done, value } = await reader.read()
        if (done) break
        const text = textDecoder.decode(value, { stream: true })
        if (text) handleText(text)
      }
    } catch (e) {
      if (active && e.name !== 'AbortError') {
        errorMsg.value = e.message || '请求失败'
      }
    } finally {
      finish()
    }
  }

  return { content, loading, errorMsg, send, stop }
}

H5 那边其实用原生 TextDecoder 就够了，它的 stream: true 参数本身就带残留字节处理能力，不需要自己写状态机。所以 #ifdef H5 分支里直接走 fetch + ReadableStream，逻辑更短也更稳。

第五步：渲染节流，别让视图跟着分片一起抖

真正跑起来之后你会发现，接口分片来得比想象的密。一次几百毫秒里能到十几个 chunk，每个 chunk 里可能就一两个字。如果每个 chunk 都直接 content.value += delta，Vue 的响应式系统会触发十几次组件更新，低端安卓机上的表现就是文字一跳一跳的，光标位置也跟着飘。

上面代码里的 scheduleFlush 就是在解决这个：所有增量先写进 pending 字符串，用一个 40ms 的定时器批量落到 content.value。40ms 大约是 25fps，人眼已经感觉不出卡顿，但渲染次数少了一个数量级。

流结束或者被中断的时候必须调一次 flushNow()，把最后那点还在 pending 里的字吐出来，否则回答会缺个尾巴。

页面里用起来是这样：

<template>
  <view class="page">
    <scroll-view
      class="chat"
      scroll-y
      :scroll-top="scrollTop"
      :scroll-with-animation="false"
    >
      <view v-for="(item, idx) in history" :key="idx" class="msg">
        <text>{{ item.content }}</text>
      </view>

      <!-- 流式中的消息单独一个节点，避免整列表重渲染 -->
      <view v-if="streaming" class="msg">
        <text>{{ content }}</text>
      </view>
    </scroll-view>

    <view class="input-bar">
      <input v-model="draft" placeholder="说点什么" />
      <button v-if="!loading" @click="onSend">发送</button>
      <button v-else @click="stop">停止</button>
    </view>
  </view>
</template>

<script setup>
import { ref, watch, nextTick, onUnmounted } from 'vue'
import { useChatStream } from '@/composables/useChatStream'

const draft = ref('')
const history = ref([])
const streaming = ref(false)
const scrollTop = ref(0)

let scrollCursor = 0

const { content, loading, errorMsg, send, stop } = useChatStream({
  url: 'https://your-api.com/v1/chat/completions',
  headers: { Authorization: 'Bearer sk-xxxx' },
  model: 'gpt-4o-mini'
})

// 内容每刷新一次，把滚动位置往下推
watch(content, () => {
  nextTick(() => {
    scrollCursor += 5000
    scrollTop.value = scrollCursor
  })
})

async function onSend() {
  const text = draft.value.trim()
  if (!text || loading.value) return

  draft.value = ''
  history.value.push({ role: 'user', content: text })

  streaming.value = true
  await send(history.value)

  history.value.push({ role: 'assistant', content: content.value })
  content.value = ''
  streaming.value = false

  if (errorMsg.value) {
    uni.showToast({ title: errorMsg.value, icon: 'none' })
  }
}

onUnmounted(() => {
  // 页面走了必须断流，否则请求会一直挂着
  stop()
})
</script>

注意模板里那个「流式消息单独一个节点」的处理。如果把正在流式输出的内容塞进 history 数组里，每次更新都会让 v-for 整段重新比对一遍，历史消息越多越卡。拆成单独节点之后，只有那一个 text 会重新渲染。

滚动那块用的是一个取巧但很管用的办法：scroll-top 每次都在上一次的值上再加 5000，保证数值一直在变，从而每次都会触发滚动到底部。如果只给一个固定的大数字，第二次赋值时值没变，scroll-view 不会响应。再配合 :scroll-with-animation="false"，就不会出现滚到一半又被新内容拽回去的抽搐感。

第六步：降级方案也得留一条

不是所有端都有 enableChunked。支付宝小程序、部分老版本基础库、还有一些 App 的旧 WebView，都不支持。这时候不能直接白屏，得有条退路。

最简单的降级是退回普通请求，等完整响应回来之后，用定时器把字符串按固定速度往外吐，模拟出逐字的效果。视觉体验差一点，但功能是完整的。

function fallbackRequest(body) {
  return new Promise((resolve) => {
    uni.request({
      url,
      method: 'POST',
      header: Object.assign({
        'Content-Type': 'application/json'
      }, headers),
      data: body,
      success: (res) => {
        const full = res.data && res.data.choices
          && res.data.choices[0]
          && res.data.choices[0].message
          && res.data.choices[0].message.content || ''

        let i = 0
        const timer = setInterval(() => {
          if (i >= full.length) {
            clearInterval(timer)
            finish()
            resolve()
            return
          }
          // 一次吐 2 个字符，比逐字更接近真实感
          content.value += full.slice(i, i + 2)
          i += 2
        }, 30)
      },
      fail: () => {
        errorMsg.value = '请求失败'
        finish()
        resolve()
      }
    })
  })
}

在 send 里判断一下 onChunkReceived 是否可用，不可用就走这条。用户基本感知不到差别，只是第一句话出现得晚一点。

真机上的几个坑

1. 开发者工具和真机行为不完全一致。 开发者工具里 onChunkReceived 的分片粒度通常更细，真机上（尤其 iOS）可能一次给你几百字节。所以测试时务必在真机上跑一遍，看有没有乱码和吞字。

2. 后端的缓冲策略会毁掉整个体验。 有些网关和 Nginx 默认会攒够一定字节才转发，导致你以为自己接的是流式，实际上是分两三批一次性到齐。服务端记得加 X-Accel-Buffering: no，响应头里也要有 Cache-Control: no-cache 和 Content-Type: text/event-stream。

3. 别在 success 里再拼一次内容。 前面提过，但还是有人踩。success 的 res.data 是整个响应的聚合结果，你已经在 onChunkReceived 里消费过了，再拼一次就是双份。

4. 页面卸载必须断流。 用户点了返回，请求还挂在那儿，回调还在往已经销毁的组件里写数据，轻则内存泄漏，重则报「Cannot read property of undefined」。onUnmounted 里调一次 stop()，task.abort() 在小程序端是有效的。

5. JSON 解析失败不要抛异常。 服务端偶尔会塞一些非标准的心跳或统计行，JSON.parse 直接 try/catch 掉继续就行。整条流因为一行脏数据崩掉是最亏的。

6. 长对话要裁剪历史。 content 会随着回答越来越长，如果多轮对话把全量历史都堆在页面上，滚动和重排都会变慢。一般保留最近 10 到 20 轮就够，更早的可以折叠或者直接丢掉。

写在最后

这套方案的核心其实就三件事：让请求别提前关闭、让解码器记住没解完的字节、让 SSE 解析器记住没切完的分片。状态机一旦立住，剩下的都是包装。

代码我尽量写得能直接落地，useChatStream 那个文件复制到项目里，改改 URL 和 header 就能跑。真正需要你自己调的通常是节流间隔（40ms 是个比较稳的值）和滚动策略，这两个跟你的页面结构关系比较大。

如果在真机上遇到什么奇怪的表现，优先怀疑两件事：服务端有没有真的在 flush，以及你的解码器有没有正确处理 carry。这两处占了流式输出里九成的疑难杂症。