ReadableStream实践

要点

  • ReadableStream 是 Web 平台原生提供的流式数据接口，用来处理「数据量大到不适合一次性加载到内存」的场景
  • 普通请求把全部数据攒好再返回，流式响应则是一块一块地往外推数据，客户端收到一块就能处理一块
  • Hono 的 c.body() 方法接受一个 ReadableStream，可以直接把它作为响应体返回
  • 创建 ReadableStream 的核心是写好 start() 和 pull() 两个回调，前者做初始化，后者负责按需提供数据
  • 流式数据在传输过程中可以用 pipeThrough() 接 TransformStream 做转换，类似管道的概念
  • 错误处理需要在 controller 里调用 controller.error()，让下游及时感知到异常
  • 背压（backpressure）是流式系统里的自我保护机制，当下游消费不过来时，上游暂停推送

1. 先从一个问题开始

假设你在做一个 AI 对话接口，用户发一句话，服务端需要调用大模型，把生成的文字逐字返回给前端，让用户看到「打字机效果」。

最直觉的写法可能是这样：

// src/index.ts
app.post('/api/chat', async (c) => {
  const result = await callLLM('你好')
  return c.json({ message: result })
})

问题在于 callLLM 需要等模型把所有内容全部生成完，才会返回。一个回答可能要生成 10 秒甚至更久，这段时间前端一直转圈，什么都没收到。

用户等了 10 秒，突然看到一整段文字蹦出来——体验不好。如果能在生成的过程中，每产出一小段就立刻推给前端，用户就能实时看到内容逐步出现。

要做到这一点，服务端需要一种「边产生边发送」的能力。ReadableStream 就是为此设计的。

2. 什么是 ReadableStream

ReadableStream 是浏览器和 Cloudflare Workers 运行时都原生支持的一个接口。它代表一个「可以分块读取」的数据源。

把它和数组做个对比：

  • 数组：所有元素在内存里，你可以随时通过下标访问任意一个
  • ReadableStream：数据不是一开始就全部存在的，而是一块一块地产生，你按顺序一块一块地读

这种设计适合几类场景：

  1. 数据量很大，不适合一次性加载到内存（比如几个 GB 的日志文件）
  2. 数据是逐步产生的（比如 AI 模型逐 token 生成）
  3. 数据来自外部持续推送（比如 WebSocket 转发、SSE 事件流）

在 Web API 里，fetch() 的 response.body 就是一个 ReadableStream。当你请求一个大文件时，浏览器不会等到整个文件下载完才给你，而是一块一块地接收和处理。

3. 创建你的第一个 ReadableStream

一个 ReadableStream 由三部分组成：

  • start()：流被创建时立刻执行，做一些初始化工作
  • pull()：当下游请求更多数据时调用，负责往流里推入新数据
  • cancel()：当下游主动关闭流时调用，用来清理资源
// src/utils/stream.ts
export function createCountStream(max: number) {
  let count = 0
 
  return new ReadableStream({
    start(controller) {
      console.log('流已启动')
    },
 
    async pull(controller) {
      count++
      if (count <= max) {
        // 往流里推入一块数据
        controller.enqueue(`第 ${count} 块数据\n`)
      } else {
        // 数据全部推完了，关闭流
        controller.close()
        console.log('流已关闭')
      }
    },
  })
}

这段代码创建了一个计数流，从 1 数到 max，每数一个就推入一块文本。数完之后调用 controller.close() 关闭流。

controller 是流的核心控制句柄，它只有三个方法：

  • controller.enqueue(data)：推入一块数据
  • controller.close()：关闭流，表示没有更多数据了
  • controller.error(err)：标记流出错，下游会收到错误

在 Hono 里把这个流作为响应返回：

// src/index.ts
import { Hono } from 'hono'
import { createCountStream } from './utils/stream'
 
const app = new Hono()
 
app.get('/api/count', (c) => {
  const stream = createCountStream(5)
 
  return c.body(stream, 200, {
    'Content-Type': 'text/plain; charset=utf-8',
  })
})
 
export default app

c.body() 是 Hono 提供的底层方法，接受一个 ReadableStream 直接作为响应体。访问 /api/count，你会看到文本一块一块地出现，而不是等所有数据攒好再返回。

4. pull 的调用时机

pull() 不是创建流时就立刻连续执行的。它的调用由下游驱动——下游每读走一块数据，pull() 才会被调用一次，推入新的数据。

这意味着：

  • 如果没人来读这个流，pull() 不会被调用，数据也不会产生
  • 如果下游读得很快，pull() 会被连续调用，直到内部缓冲区填满
  • 如果下游读得很慢，pull() 会暂停，等下游消费了再继续

这种机制叫做「按需拉取」（lazy pull），是流式系统的基础设计。它保证上游不会无脑地往内存里灌数据——下游不读，上游就不产生，内存占用始终可控。

在 AI 对话场景里，这种机制的好处尤其明显：模型生成一个 token，pull() 被调用一次，推入一块数据。模型还没生成完的部分不会占内存，客户端还没接收的部分也不会堆积。

5. 在 Hono 中实现流式 AI 响应

回到开头提到的 AI 对话场景。调用大模型 API 时，大多数服务商（OpenAI、Anthropic 等）都支持流式返回，返回的 response.body 本身就是一个 ReadableStream。

最直接的做法是把上游的流直接透传给客户端：

// src/routes/chat.ts
import { Hono } from 'hono'
 
const app = new Hono()
 
app.post('/api/chat', async (c) => {
  const { message } = await c.req.json()
 
  // 调用大模型 API，开启流式返回
  const upstream = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${c.env.OPENAI_API_KEY}`,
    },
    body: JSON.stringify({
      model: 'gpt-4o',
      messages: [{ role: 'user', content: message }],
      stream: true,
    }),
  })
 
  // upstream.body 就是 ReadableStream，直接透传
  return c.body(upstream.body, 200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
  })
})
 
export default app

fetch() 返回的 response.body 是 ReadableStream，直接丢给 c.body() 就完成了透传。数据从 OpenAI 的服务器流过你的 Worker，直接到达客户端，不需要在中间做缓存。

6. 用 TransformStream 转换流数据

直接透传虽然简单，但上游返回的格式不一定适合你的前端。比如 OpenAI 的流式响应是 Server-Sent Events（SSE）格式，每一行长这样：

data: {"choices":[{"delta":{"content":"你"}}]}
 
data: {"choices":[{"delta":{"content":"好"}}]}
 
data: [DONE]

前端拿到的是一大坨 JSON 字符串，解析起来不方便。也许你想在服务端把它转成更简洁的纯文本流，只吐出每个 token 的文字内容。

这时候可以用 TransformStream 在中间做转换：

// src/utils/transform.ts
export function createTokenTransform() {
  let buffer = ''
 
  return new TransformStream<string, string>({
    transform(chunk, controller) {
      buffer += chunk
 
      // SSE 格式按换行分隔，每行以 "data: " 开头
      const lines = buffer.split('\n')
      // 最后一行可能不完整，留到下次处理
      buffer = lines.pop() ?? ''
 
      for (const line of lines) {
        const trimmed = line.trim()
        if (!trimmed.startsWith('data: ')) continue
 
        const data = trimmed.slice(6)
        if (data === '[DONE]') {
          controller.terminate()
          return
        }
 
        try {
          const parsed = JSON.parse(data)
          const token = parsed.choices?.[0]?.delta?.content
          if (token) {
            controller.enqueue(token)
          }
        } catch {
          // 格式异常的跳过
        }
      }
    },
  })
}

然后在路由里用 pipeThrough() 把流接上这个转换。接法和第 5 节的透传类似，区别只是在中间多套了两层 pipeThrough()：

// src/routes/chat.ts
// ... 前面的 fetch 代码和第 5 节一样 ...
 
if (!upstream.body) {
  return c.json({ error: '上游未返回流' }, 502)
}
 
// ReadableStream -> TextDecoderStream -> TransformStream -> 纯文本流
const textStream = upstream.body
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(createTokenTransform())
 
return c.body(textStream, 200, {
  'Content-Type': 'text/plain; charset=utf-8',
})

pipeThrough() 的用法类似 Unix 管道：上游输出接下游输入，数据流过中间每一层时被逐步转换。

这条数据流的完整路径是：

OpenAI 服务器 -> upstream.body (ReadableStream<Uint8Array>)
  -> TextDecoderStream (转成 string)
  -> createTokenTransform (提取 token 文本)
  -> c.body() (作为 HTTP 响应返回给客户端)

客户端收到的就是纯文本的 token 流，一个汉字一个汉字地往外蹦，不需要再解析 SSE 格式。

7. 错误处理

流式传输过程中随时可能出问题：上游 API 超时、网络抖动、模型返回异常格式。和普通请求不同，流式响应的错误不一定发生在请求发起时，可能在传了几十块数据之后才出现。

处理流中的错误，核心是在 pull() 或 transform() 里调用 controller.error()：

// src/utils/safe-stream.ts
export function createSafeStream(fetchPromise: Promise<Response>) {
  return new ReadableStream({
    async start(controller) {
      try {
        const response = await fetchPromise
        if (!response.body) {
          controller.error(new Error('上游未返回响应体'))
          return
        }
 
        const reader = response.body.getReader()
        while (true) {
          const { done, value } = await reader.read()
          if (done) {
            controller.close()
            break
          }
          controller.enqueue(value)
        }
      } catch (err) {
        controller.error(new Error(`流传输中断: ${err}`))
      }
    },
  })
}

controller.error() 和 controller.close() 的区别：

  • close()：正常结束，下游会收到流结束的信号
  • error()：异常结束，下游的 reader.read() 会抛出错误

在路由层，用 try/catch 可以兜底 fetch() 发起阶段的错误，以及检查上游返回的状态码：

// src/routes/chat.ts
try {
  const upstream = await fetch(...)
 
  if (!upstream.ok) {
    return c.json({ error: '上游 API 报错' }, 502)
  }
 
  return c.body(upstream.body)
} catch (err) {
  return c.json({ error: '服务内部错误' }, 500)
}

但要注意：try/catch 只能抓到 fetch() 发起阶段的错误。流已经开始传输之后出的错，需要在前面的 controller.error() 里处理。

8. 背压：下游消费不过来时怎么办

前面提到 pull() 是按需调用的——下游读走一块，上游才推入一块。这种机制本身就是一种背压（backpressure）控制。

但有些场景需要你自己处理背压。比如你往流里写数据的速度比下游读走的速度快，数据会堆积在流的内部缓冲区里。ReadableStream 有一个 highWaterMark 参数，用来设置内部缓冲区的上限：

// src/utils/buffered-stream.ts
export function createBufferedStream() {
  let index = 0
 
  return new ReadableStream(
    {
      async pull(controller) {
        index++
        // 模拟大量数据产生
        const chunk = 'x'.repeat(1024 * 64) // 每块 64KB
        controller.enqueue(chunk)
 
        if (index >= 100) {
          controller.close()
        }
      },
    },
    {
      // 内部缓冲区最多缓存 3 块数据
      highWaterMark: 3,
    }
  )
}

当缓冲区里有 3 块数据还没被读走时，pull() 就不会再被调用，上游暂停产生数据。等下游消费了一块，缓冲区降到 2 块，pull() 才会再次触发。

这样内存不会无限增长——不管数据总量多大，同一时刻内存里最多只缓存 highWaterMark + 1 块数据。

对于前面 AI 对话的例子，背压通常是自动处理的。大模型生成 token 的速度和客户端接收的速度大致匹配，不需要手动调整。但如果你在做文件上传转发、日志流聚合这类场景，highWaterMark 是一个重要的调优参数。

9. 用 ReadableStream 手动构建 SSE

前面用 pipeThrough 解析上游的 SSE，但如果你想自己作为 SSE 服务端，向前端推送事件呢？

ReadableStream 也能做到：

// src/routes/events.ts
import { Hono } from 'hono'
 
const app = new Hono()
 
app.get('/api/events', (c) => {
  let index = 0
 
  const stream = new ReadableStream({
    async pull(controller) {
      index++
 
      // SSE 格式：data: 内容\n\n
      const event = `data: ${JSON.stringify({ index, time: Date.now() })}\n\n`
      controller.enqueue(event)
 
      // 每秒推一条
      await new Promise((resolve) => setTimeout(resolve, 1000))
 
      if (index >= 10) {
        controller.close()
      }
    },
  })
 
  return c.body(stream, 200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
  })
})
 
export default app

客户端用 EventSource 接收即可：

// client.ts
const source = new EventSource('/api/events')
source.onmessage = (event) => {
  const data = JSON.parse(event.data)
  console.log(`收到第 ${data.index} 条，时间戳：${data.time}`)
}

这样服务端每 1 秒推一条事件，客户端实时收到。比轮询高效得多，也比 WebSocket 轻量。

总结

回顾一下这篇的内容：

  • ReadableStream 把数据按块产生、按块读取，适合大数据量和逐步生成的场景
  • 创建流时写好 start() 做初始化、pull() 按需提供数据、cancel() 清理资源
  • Hono 的 c.body() 直接接受 ReadableStream 作为响应体
  • pipeThrough() 接 TransformStream 可以在传输过程中转换数据格式
  • 错误处理用 controller.error() 让下游感知异常，普通 try/catch 只能抓到请求发起阶段的错误
  • 背压由 pull() 的按需调用机制自动实现，highWaterMark 控制缓冲上限

流式处理在 AI 接口、大文件传输、实时推送这类场景下几乎是标配。ReadableStream 作为 Web 平台原生接口，不需要额外的库，在 Cloudflare Workers 上直接可用。