ReadableStream reader

ReadableStream 把生产者和消费者解耦：生产者按 chunk 入队，reader 逐次 read()，直到收到 { done: true }。chunk 只是传输边界，不能直接当成业务消息；文本协议必须自己处理残片和分隔符。

一个不依赖网络的实验

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue(new TextEncoder().encode('hel'))
    controller.enqueue(new TextEncoder().encode('lo\nwo'))
    controller.enqueue(new TextEncoder().encode('rld\n'))
    controller.close()
  },
  cancel(reason) { console.log('cancelled:', reason) }
})

const reader = stream.getReader()
const decoder = new TextDecoder()
let buffer = ''
while (true) {
  const { value, done } = await reader.read()
  if (done) break
  buffer += decoder.decode(value, { stream: true })
  const lines = buffer.split('\n')
  buffer = lines.pop()
  for (const line of lines) console.log('line:', line)
}
buffer += decoder.decode()
if (buffer) console.log('truncated:', buffer)

日志应依次为 hel 与 chunk 无关的完整行 hello、world。把最后一个 chunk 改成 rld，结束时应出现 truncated: hello\nworld 的残片，而不是悄悄丢失。

读取、锁与取消

调用 getReader() 会锁定流；另一个消费者不能同时读取。路由离开或用户点击停止时调用 reader.cancel(reason)，并在 finally 中 releaseLock()。网络 fetch 的 response.body 可能为 null（例如无响应体或宿主不支持），调用前要先判断。

背压与宿主边界

慢消费者会让队列增长，生产者应根据 desiredSize 或 pull() 时机减速。浏览器页面隐藏、Service Worker 和 Node 的 stream 实现有差异；本实验只验证 Web Streams API 的内存对象，不等价于真实网络断线。

练习

1. 让消费者每 100ms 处理一行，记录队列变化并解释背压。
2. 中途调用 cancel('route-left')，验证生产者的 cancel 回调和最终日志。
3. 将输入改为多字节 UTF-8 字符，说明为什么要复用同一个 TextDecoder。

查看本节答案