> ## Documentation Index > Fetch the complete documentation index at: https://bun.zhcndoc.com/llms.txt > Use this file to discover all available pages before exploring further. # 流 > 使用 Bun 的流 API 以无需一次性将全部数据加载到内存中即可处理二进制数据流是处理二进制数据而无需一次性全部加载到内存中的重要抽象。它们通常用于读写文件、发送和接收网络请求以及处理大量数据。 Bun 实现了 Web API 中的 [`ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream) 和 [`WritableStream`](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream)。 Bun 也实现了 `node:stream` 模块，包含 [`Readable`](https://nodejs.org/api/stream.html#stream_readable_streams)、 [`Writable`](https://nodejs.org/api/stream.html#stream_writable_streams) 和 [`Duplex`](https://nodejs.org/api/stream.html#stream_duplex_and_transform_streams)。更多完整文档请参考 [Node.js 文档](https://nodejs.org/api/stream.html)。创建一个 `ReadableStream`： ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const stream = new ReadableStream({ start(controller) { controller.enqueue("hello"); controller.enqueue("world"); controller.close(); }, }); ``` 可以使用 `for await` 语法逐块读取 `ReadableStream` 的内容。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} for await (const chunk of stream) { console.log(chunk); } // hello // world ``` *** ## 直接 `ReadableStream` Bun 实现了一个优化版本的 `ReadableStream`，避免了不必要的数据复制和队列管理逻辑。在传统的 `ReadableStream` 中，数据块是被 *入队* 的。每个数据块都会被复制到一个队列中，直到流准备好发送更多数据时才会被读取。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const stream = new ReadableStream({ start(controller) { controller.enqueue("hello"); controller.enqueue("world"); controller.close(); }, }); ``` 而直接 `ReadableStream` 中，数据块会直接写入流中。不会有排队，也无需将数据块复制到内存中。`controller` 的 API 也做了相应更新，不使用 `.enqueue()`，而是调用 `.write`。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const stream = new ReadableStream({ type: "direct", // [!code ++] pull(controller) { controller.write("hello"); controller.write("world"); }, }); ``` 当使用直接 `ReadableStream` 时，目标端会处理所有的数据块队列。流的消费者将收到与传给 `controller.write()` 完全一致的内容，不进行任何编码或修改。 ### 处理背压 `controller.write()` 返回已写入的字节数。当目标端的内部缓冲区已满时（例如较慢的 HTTP 客户端），它会改为返回一个**负数**。数据块仍然会被接受——负返回值表示需要暂停并等待目标端将数据排空。要等待排空，请 `await controller.flush(true)`： ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const stream = new ReadableStream({ type: "direct", async pull(controller) { for (const chunk of chunks) { const n = controller.write(chunk); if (typeof n === "number" && n < 0) { // 目标端已积压；在写入更多数据之前等待其排空 await controller.flush(true); } } controller.close(); }, }); ``` 对于默认（非 `direct`）的 `ReadableStream` 和异步生成器响应体，Bun 会自动应用这种背压——当目标端积压时，生产者会暂停。 *** ## 异步生成器流 Bun 也支持将异步生成器函数作为 `Response` 和 `Request` 的数据源。使用异步生成器创建一个从异步源获取数据的 `ReadableStream`。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const response = new Response( (async function* () { yield "hello"; yield "world"; })(), ); await response.text(); // "helloworld" ``` 你也可以直接使用 `[Symbol.asyncIterator]`。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const response = new Response({ [Symbol.asyncIterator]: async function* () { yield "hello"; yield "world"; }, }); await response.text(); // "helloworld" ``` 如果你需要对流进行更细粒度的控制，`yield` 将返回直接的 ReadableStream 控制器。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const response = new Response({ [Symbol.asyncIterator]: async function* () { const controller = yield "hello"; await controller.end(); }, }); await response.text(); // "hello" ``` *** ## `Bun.ArrayBufferSink` `Bun.ArrayBufferSink` 类是一个用于构造未知大小 `ArrayBuffer` 的高速增量写入器。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} 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` 选项。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const sink = new Bun.ArrayBufferSink(); sink.start({ asUint8Array: true, // [!code ++] }); 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`。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} sink.write("h"); sink.write(new Uint8Array([101, 108])); sink.write(Buffer.from("lo").buffer); sink.end(); ``` 一旦调用 `.end()`，就无法再向 `ArrayBufferSink` 写入数据。然而，在流缓冲的场景下，通常需要持续写入数据并定期调用 `.flush()` 将内容（例如写入到一个 `WritableStream`）。为支持此需求，可在构造时传入 `stream: true`。 ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const sink = new Bun.ArrayBufferSink(); sink.start({ stream: true, // [!code ++] }); sink.write("h"); sink.write("e"); sink.write("l"); sink.flush(); // ArrayBuffer(5) [ 108, 101, 108 ] sink.write("l"); sink.write("o"); sink.flush(); // ArrayBuffer(5) [ 108, 111 ] ``` `.flush()` 方法返回缓冲数据（作为 `ArrayBuffer`，如果 `asUint8Array: true` 则返回 `Uint8Array`）并清空内部缓冲区。如果需要手动设置内部缓冲区的大小（以字节为单位），请传入 `highWaterMark`： ```ts theme={"theme":{"light":"github-light","dark":"dracula"}} const sink = new Bun.ArrayBufferSink(); sink.start({ highWaterMark: 1024 * 1024, // 1 MB // [!code ++] }); ``` *** ## 参考 ```ts See Typescript Definitions expandable theme={"theme":{"light":"github-light","dark":"dracula"}} /** * 快速增量写入器，调用 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; end(): ArrayBuffer | Uint8Array; } ```