Stream API

ReadableStream

表示数据源，允许你逐步读取数据块，常用于处理网络响应、文件读取、自定义数据生成等。

核心特性：

• 数据只能被消费一次。

• 支持背压（backpressure）：消费者控制读取速度。

• 可通过getReader()获取ReadableStreamDefaultReader

• 支持异步迭代（for await...of）。

创建方式

• 使用构建函数（自定义流）

const readable = new ReadableStream({
  start(controller) {
    controller.enqueue('chunk 1');
    controller.enqueue('chunk 2');
    controller.close(); // 结束流
  }
});

• 来自 Fetch API

const response = await fetch('/data');
const readable = response.body; // ReadableStream

读取方式

• 使用 reader

const reader = readable.getReader();
const { done, value } = await reader.read();

• 使用异步迭代器（推荐）

for await (const chunk of readable) {
  console.log(chunk);
}

一旦使用 getReader() 或 pipeTo()，流即被锁定（locked），不能被其他方式读取。

WritableStream（可写流）

表述数据目的地，允许你逐步写入数据块，如上床到服务器，写入文件等。

核心特性：

• 支持背压：写入时可等待“准备好”信号。

• 可通过getWriter() 获取 WritableStreamDefaultWriter。

创建方式

const writable = new WritableStream({
  write(chunk) {
    console.log('Writing:', chunk);
  },
  close() {
    console.log('Stream closed');
  },
  abort(reason) {
    console.error('Stream aborted:', reason);
  }
});

写入方式

• 使用 writer

const writer = writable.getWriter();
await writer.write('Hello');
await writer.close();

• 通过管道接收数据

readableStream.pipeTo(writableStream);

TransformStream（转换流）

同时包含一个 WritableStream（输入） 和一个 ReadableStream（输出），用于中间处理数据（如压缩、加密、格式转换）。

核心特性

• 输入写入 → 转换逻辑 → 输出可读。

• 常用于 .pipeThrough() 链式处理。

创建方式

const transform = new TransformStream({
  transform(chunk, controller) {
    // 处理每个 chunk
    const upper = chunk.toString().toUpperCase();
    controller.enqueue(upper); // 输出到 readable 端
  },
  flush(controller) {
    // 所有输入处理完毕后调用
    controller.enqueue('[END]');
  }
});

使用方式：管道转换

// readable → transform → newReadable
const newReadable = readableStream.pipeThrough(transform);

// 或直接连接到可写流
readableStream
  .pipeThrough(transform)
  .pipeTo(writableStream);

关键概念补充

概念	说明
背压（Backpressure）	流自动协调生产者和消费者速度，防止内存溢出。
流锁定（Locked）	一旦流被 reader/writer/pipe 占用，就不能再被其他方式访问。
流不可重用	流是一次性的，关闭或取消后不能重新开始。
异步迭代支持	所有 ReadableStream 都实现 [Symbol.asyncIterator]。