流 - Bun 中文文档

流是处理二进制数据而无需一次性全部加载到内存中的重要抽象。它们通常用于读写文件、发送和接收网络请求以及处理大量数据。 Bun 实现了 Web API 中的 ReadableStream 和 WritableStream。

Bun 也实现了 node:stream 模块，包含 Readable、 Writable 和 Duplex。更多完整文档请参考 Node.js 文档。

创建一个 ReadableStream：

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue("hello");
    controller.enqueue("world");
    controller.close();
  },
});

你可以使用 for await 语法逐块读取 ReadableStream 的内容。

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

// 你好
// 世界

直接 `ReadableStream`

Bun 实现了优化版的 ReadableStream，避免了不必要的队列管理。使用传统的 ReadableStream 时，你需要将数据块入队。流会将每个数据块添加到队列中，直到流准备好发送更多数据时，这些数据块才会留在队列中。

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue("hello");
    controller.enqueue("world");
    controller.close();
  },
});

使用直接 ReadableStream 时，你可以直接将数据块写入流中，不会发生排队。controller API 也体现了这一点：你调用 .write() 而不是 .enqueue()。

const stream = new ReadableStream({
  type: "direct", 
  pull(controller) {
    controller.write("hello");
    controller.write("world");
    controller.close();
  },
});

使用直接 ReadableStream 时，目标端会处理所有数据块的排队。目标端会接收你传递给 controller.write() 的字节。从 JavaScript 读取流时，Bun 会缓冲写入内容，并以 Uint8Array 数据块的形式传递（字符串会进行 UTF-8 编码）。

结束流

所有内容写入完毕后，调用 controller.close()（或 controller.end()）。每个目标端都会刷新已缓冲的内容，并在此时结束。controller.close(error) 则会使流失败：不会再刷新任何内容，消费者会因 error 而拒绝。接收整个消息体的目标端（Bun.serve、fetch、Bun.write、Bun.spawn stdin、.text() 等）会调用一次 pull()。如果它返回一个 promise，那么在该 promise 处于待定状态期间，流会保持打开；当 promise 兑现时，流会结束，效果相当于 pull() 最后调用了 controller.close()；如果 promise 被拒绝，流会因该原因而出错。pull() 若不返回 promise 且不关闭流，则流会保持打开，直到有内容调用 controller.close()，因此你可以保留 controller，并从事件中写入内容。通过 reader（getReader()、for await、pipeTo()、tee()）读取流时，pull() 则是一种需求信号：在之前调用返回的 promise 仍处于待定状态时不会再次调用，而是会在后续读取时再次调用，直到 controller.close()。

处理背压

controller.write() 会返回已写入的字节数；如果目标端的内部缓冲区已满（例如，HTTP 客户端速度较慢），则会返回一个待定的 Promise<number>。无论哪种情况，数据块都会被接收。目标端断开连接后（例如，HTTP 客户端断开连接），write() 会返回 0。当目标端缓冲区排空后，promise 就会兑现，因此只需对结果使用 await：

const stream = new ReadableStream({
  type: "direct",
  async pull(controller) {
    for (const chunk of chunks) {
      await controller.write(chunk);
    }
    controller.close();
  },
});

await controller.flush(true) 效果相同，也可以在写入返回 Promise 后使用它。从 JavaScript 读取流时（getReader()、pipeTo()、for await），当有 highWaterMark 字节（默认为 64 KiB）等待 reader 读取时，缓冲区就会被视为已满。将 { highWaterMark } 作为 ReadableStream 构造函数的第二个参数传入，即可更改此值。reader.cancel(reason) 会调用源的 cancel(reason)。对于默认（非 direct）的 ReadableStream 和异步生成器响应体，Bun 会自动应用此背压：当目标端积压时，它会暂停生产者。

异步生成器流

Bun 也支持将异步生成器函数作为 Response 和 Request 的数据源。使用异步生成器创建一个从异步源获取数据的 ReadableStream。

const response = new Response(
  (async function* () {
    yield "hello";
    yield "world";
  })(),
);

await response.text(); // "helloworld"

你也可以直接使用 [Symbol.asyncIterator]。

const response = new Response({
  [Symbol.asyncIterator]: async function* () {
    yield "hello";
    yield "world";
  },
});

await response.text(); // "helloworld"

消息体只包含生成器 yield 的值。与 for await 一样，生成器的返回值不属于消息体。

const response = new Response(
  (async function* () {
    yield "hello";
    return "ignored";
  })(),
);

await response.text(); // "hello"

如需更精细地控制流，yield 会返回直接 ReadableStream 的 controller。

const response = new Response({
  [Symbol.asyncIterator]: async function* () {
    const controller = yield "hello";
    await controller.end();
  },
});

await response.text(); // "hello"

`Bun.ArrayBufferSink`

Bun.ArrayBufferSink 类是一个用于构造未知大小 ArrayBuffer 的高速增量写入器。

const sink = new Bun.ArrayBufferSink();

sink.write("h");
sink.write("e");
sink.write("l");
sink.write("l");
sink.write("o");

sink.end();
// ArrayBuffer(5) [ 104, 101, 108, 108, 111 ]

如果需要以 Uint8Array 格式获取数据，可以向 start 方法传入 asUint8Array 选项。

const sink = new Bun.ArrayBufferSink();
sink.start({
  asUint8Array: true, 
});

sink.write("h");
sink.write("e");
sink.write("l");
sink.write("l");
sink.write("o");

sink.end();
// Uint8Array(5) [ 104, 101, 108, 108, 111 ]

.write() 方法支持字符串、类型化数组、ArrayBuffer 和 SharedArrayBuffer。

sink.write("h");
sink.write(new Uint8Array([101, 108]));
sink.write(Buffer.from("lo").buffer);

sink.end();

调用 .end() 后，就不能再向 ArrayBufferSink 写入数据。不过，在缓冲流时，你可能希望继续写入数据，并定期对内容调用 .flush()（例如，写入 WritableStream）。为此，请向 start 方法传入 stream: true。

const sink = new Bun.ArrayBufferSink();
sink.start({
  stream: true, 
});

sink.write("h");
sink.write("e");
sink.write("l");
sink.flush();
// ArrayBuffer(3) [ 104, 101, 108 ]

sink.write("l");
sink.write("o");
sink.flush();
// ArrayBuffer(2) [ 108, 111 ]

.flush() 方法会将缓冲的数据作为 ArrayBuffer 返回（如果设置了 asUint8Array: true，则返回 Uint8Array），并清空内部缓冲区。如果需要手动设置内部缓冲区的大小（以字节为单位），请传入 highWaterMark：

const sink = new Bun.ArrayBufferSink();
sink.start({
  highWaterMark: 1024 * 1024, // 1 MB
});

参考

查看可展开的 TypeScript 定义

/**
 * 快速增量写入器，调用 end() 后生成 `ArrayBuffer`。
 */
export class ArrayBufferSink {
  constructor();

  start(options?: {
    asUint8Array?: boolean;
    /**
     * 预分配此大小的内部缓冲区
     * 当数据块较小时，能显著提升性能
     */
    highWaterMark?: number;
    /**
     * 在 {@link ArrayBufferSink.flush} 时返回写入数据为 `Uint8Array`。
     * 写入将从缓冲区开头重新开始。
     */
    stream?: boolean;
  }): void;

  write(chunk: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer): number;
  /**
   * 刷新内部缓冲区
   *
   * 如果 {@link ArrayBufferSink.start} 传入了 `stream` 选项，则返回一个 `ArrayBuffer`
   * 若同时传入了 `stream` 和 `asUint8Array`，则返回 `Uint8Array`
   * 否则，返回自上次刷新以来写入的字节数
   *
   * 该 API 以后可能拆分为 Uint8ArraySink 和 ArrayBufferSink
   */
  flush(): number | Uint8Array<ArrayBuffer> | ArrayBuffer;
  end(): ArrayBuffer | Uint8Array<ArrayBuffer>;
}

开始使用

核心运行时

文件与模块系统

HTTP 服务器

网络通信

数据与存储

并发

进程与系统

交互与工具链

实用工具

标准与兼容性

贡献

流

直接 `ReadableStream`

结束流

处理背压

异步生成器流

`Bun.ArrayBufferSink`

参考

​直接 ReadableStream

​结束流

​处理背压

​异步生成器流

​Bun.ArrayBufferSink

​参考

直接 `ReadableStream`

结束流

处理背压

异步生成器流

`Bun.ArrayBufferSink`

参考