Node.js Shell 脚本开发指南（中）

九、原生 Node.js 流

原文：exploringjs.com/nodejs-shell-scripting/ch_nodejs-streams.html

译者：飞龙

协议：CC BY-NC-SA 4.0

• 9.1 总结：异步迭代和异步生成器

• 9.2 流

  • 9.2.1 管道

  • 9.2.2 文本编码

  • 9.2.3 辅助函数：readableToString()

  • 9.2.4 一些初步说明

• 9.3 可读流

  • 9.3.1 创建可读流

  • 9.3.2 通过for-await-of从可读流中读取块

  • 9.3.3 通过模块'node:readlines'从可读流中读取行

• 9.4 通过异步生成器转换可读流(ch_nodejs-streams.html#transforming-Readable-via-async-generator)

  • 9.4.1 从异步可迭代对象中的块转换为编号行

• 9.5 可写流

  • 9.5.1 为文件创建可写流

  • 9.5.2 写入可写流

• 9.6 快速参考：与流相关的功能(ch_nodejs-streams.html#quick-reference-stream-related-functionality)

• 9.7 进一步阅读和本章的来源

本章是对 Node 的原生流的介绍。它们支持异步迭代，这使它们更容易使用，这也是我们在本章中主要使用的。

请注意，跨平台的web 流在§10“在 Node.js 上使用 web 流”中有所涵盖。我们在本书中主要使用这些。因此，如果您愿意，可以跳过当前章节。

9.1 总结：异步迭代和异步生成器

异步迭代是一种异步检索数据容器内容的协议（意味着当前的“任务”在检索项目之前可能会暂停）。

异步生成器有助于异步迭代。例如，这是一个异步生成器函数：

/**
 * @returns an asynchronous iterable
 */
async function* asyncGenerator(asyncIterable) {
 for await (const item of asyncIterable) { // input
 if (···) {
 yield '> ' + item; // output
 }
 }
}

• for-await-of循环遍历输入的asyncIterable。这个循环也适用于普通的异步函数。

• yield将值提供给此生成器返回的异步可迭代对象。

在本章的其余部分，请仔细注意函数是异步函数还是异步生成器函数：

/** @returns a Promise */
async function asyncFunction() { /*···*/ }

/** @returns an async iterable */
async function* asyncGeneratorFunction() { /*···*/ }

9.2 流

流是一种模式，其核心思想是“分而治之”大量数据：如果我们将其分割成较小的部分并一次处理一部分，我们就可以处理它。

Node.js 支持几种流，例如：

• 可读流是我们可以从中读取数据的流。换句话说，它们是数据的来源。一个例子是可读文件流，它允许我们读取文件的内容。

• 可写流是我们可以写入数据的流。换句话说，它们是数据的接收端。一个例子是可写文件流，它允许我们向文件写入数据。

• 转换流既可读又可写。作为可写流，它接收数据块，转换（更改或丢弃）它们，然后将它们作为可读流输出。

9.2.1 管道

为了在多个步骤中处理流数据，我们可以管道（连接）流：

1. 输入通过可读流接收。

2. 每个处理步骤都是通过转换流执行的。

3. 对于最后的处理步骤，我们有两个选项：

   • 我们可以将最近的可读流中的数据写入可写流。也就是说，可写流是我们管道的最后一个元素。

   • 我们可以以其他方式处理最近的可读流中的数据。

部分（2）是可选的。

9.2.2 文本编码

创建文本流时，最好始终指定编码：

• Node.js 文档中有支持的编码及其默认拼写的列表 - 例如：

  • ‘utf8’

  • ‘utf16le’

  • ‘base64’

• 也允许一些不同的拼写。您可以使用Buffer.isEncoding()来检查哪些是：

  > buffer.Buffer.isEncoding('utf8')
  true
  > buffer.Buffer.isEncoding('utf-8')
  true
  > buffer.Buffer.isEncoding('UTF-8')
  true
  > buffer.Buffer.isEncoding('UTF:8')
  false
  

编码的默认值是null，等同于'utf8'。

9.2.3 辅助函数：readableToString()

我们偶尔会使用以下辅助函数。您不需要理解它的工作原理，只需（大致）了解它的作用。

import * as stream from 'stream';

/**
 * Reads all the text in a readable stream and returns it as a string,
 * via a Promise.
 * @param  {stream.Readable} readable
 */
function readableToString(readable) {
 return new Promise((resolve, reject) => {
 let data = '';
 readable.on('data', function (chunk) {
 data += chunk;
 });
 readable.on('end', function () {
 resolve(data);
 });
 readable.on('error', function (err) {
 reject(err);
 });
 });
}

此函数是通过基于事件的 API 实现的。稍后我们将看到一个更简单的方法 - 通过异步迭代。

9.2.4 一些初步说明

• 在本章中，我们只会使用文本流。

• 在示例中，我们偶尔会遇到await被用于顶层。在这种情况下，我们假设我们在模块内或在异步函数的主体内。

• 每当有换行符时，我们都支持：

  • Unix：'\n'（LF）

  • Windows：'\r\n'（CR LF）当前平台的换行符可以通过模块os中的常量EOL访问。

9.3 可读流

9.3.1 创建可读流

9.3.1.1 从文件创建可读流

我们可以使用fs.createReadStream()来创建可读流：

import * as fs from 'fs';

const readableStream = fs.createReadStream(
 'tmp/test.txt', {encoding: 'utf8'});

assert.equal(
 await readableToString(readableStream),
 'This is a test!\n');

9.3.1.2 Readable.from(): 从可迭代对象创建可读流

静态方法Readable.from(iterable, options?)创建一个可读流，其中包含iterable中包含的数据。iterable可以是同步可迭代对象或异步可迭代对象。参数options是可选的，可以用于指定文本编码等其他内容。

import * as stream from 'stream';

function* gen() {
 yield 'One line\n';
 yield 'Another line\n';
}
const readableStream = stream.Readable.from(gen(), {encoding: 'utf8'});
assert.equal(
 await readableToString(readableStream),
 'One line\nAnother line\n');

9.3.1.2.1 从字符串创建可读流

Readable.from()接受任何可迭代对象，因此也可以用于将字符串转换为流：

import {Readable} from 'stream';

const str = 'Some text!';
const readable = Readable.from(str, {encoding: 'utf8'});
assert.equal(
 await readableToString(readable),
 'Some text!');

目前，Readable.from()将字符串视为任何其他可迭代对象，因此会迭代其代码点。从性能上讲，这并不理想，但对于大多数用例来说应该是可以的。我期望Readable.from()经常与字符串一起使用，所以也许将来会有优化。

9.3.2 通过for-await-of从可读流中读取块

每个可读流都是异步可迭代的，这意味着我们可以使用for-await-of循环来读取其内容：

import * as fs from 'fs';

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

const readable = fs.createReadStream(
 'tmp/test.txt', {encoding: 'utf8'});
logChunks(readable);

// Output:
// 'This is a test!\n'

9.3.2.1 在字符串中收集可读流的内容

以下函数是本章开头所见函数的简化重新实现。

import {Readable} from 'stream';

async function readableToString2(readable) {
 let result = '';
 for await (const chunk of readable) {
 result += chunk;
 }
 return result;
}

const readable = Readable.from('Good morning!', {encoding: 'utf8'});
assert.equal(await readableToString2(readable), 'Good morning!');

请注意，在这种情况下，我们必须使用异步函数，因为我们想要返回一个 Promise。

9.3.3 通过模块'node:readlines'从可读流中读取行

内置模块'node:readline'让我们可以从可读流中读取行：

import * as fs from 'node:fs';
import * as readline from 'node:readline/promises';

const filePath = process.argv[2]; // first command line argument

const rl = readline.createInterface({
 input: fs.createReadStream(filePath, {encoding: 'utf-8'}),
});
for await (const line of rl) {
 console.log('>', line);
}
rl.close();

9.4 通过异步生成器转换可读流

异步迭代提供了一个优雅的替代方案，用于在多个步骤中处理流式数据的转换流：

• 输入是一个可读流。

• 第一个转换是通过一个异步生成器执行的，该生成器遍历可读流并在适当时产生。

• 可选地，我们可以通过使用更多的异步生成器来进一步转换。

• 最后，我们有几种处理最后一个生成器返回的异步可迭代对象的选项：

  • 我们可以通过Readable.from()将其转换为可读流（稍后可以传输到可写流）。

  • 我们可以使用异步函数来处理它。

  • 等等。

总之，这些是这样的处理管道的组成部分：

可读的

→ 第一个异步生成器 [→ … → 最后一个异步生成器]

→ 可读或异步函数

9.4.1 从块到异步可迭代对象中的编号行

在下一个示例中，我们将看到一个刚刚解释过的处理管道的示例。

import {Readable} from 'stream';

/**
 * @param  chunkIterable An asynchronous or synchronous iterable
 * over “chunks” (arbitrary strings)
 * @returns An asynchronous iterable over “lines”
 * (strings with at most one newline that always appears at the end)
 */
async function* chunksToLines(chunkIterable) {
 let previous = '';
 for await (const chunk of chunkIterable) {
 let startSearch = previous.length;
 previous += chunk;
 while (true) {
 // Works for EOL === '\n' and EOL === '\r\n'
 const eolIndex = previous.indexOf('\n', startSearch);
 if (eolIndex < 0) break;
 // Line includes the EOL
 const line = previous.slice(0, eolIndex+1);
 yield line;
 previous = previous.slice(eolIndex+1);
 startSearch = 0;
 }
 }
 if (previous.length > 0) {
 yield previous;
 }
}

async function* numberLines(lineIterable) {
 let lineNumber = 1;
 for await (const line of lineIterable) {
 yield lineNumber + ' ' + line;
 lineNumber++;
 }
}

async function logLines(lineIterable) {
 for await (const line of lineIterable) {
 console.log(line);
 }
}

const chunks = Readable.from(
 'Text with\nmultiple\nlines.\n',
 {encoding: 'utf8'});
await logLines(numberLines(chunksToLines(chunks))); // (A)

// Output:
// '1 Text with\n'
// '2 multiple\n'
// '3 lines.\n'

处理管道在 A 行设置。步骤是：

• chunksToLines(): 从具有块的异步可迭代对象转换为具有行的异步可迭代对象。

• numberLines(): 从具有行的异步可迭代对象转换为具有编号行的异步可迭代对象。

• logLines(): 记录异步可迭代对象中的项目。

观察：

• chunksToLines()和numberLines()的输入和输出都是异步可迭代对象。这就是为什么它们是异步生成器（由async和*指示）。

• logLines()的输入是异步可迭代对象。这就是为什么它是一个异步函数（由async指示）。

9.5 可写流

9.5.1 创建文件的可写流

我们可以使用fs.createWriteStream()来创建可写流：

const writableStream = fs.createWriteStream(
 'tmp/log.txt', {encoding: 'utf8'});

9.5.2 向可写流写入数据

在本节中，我们将探讨向可写流写入数据的方法：

1. 通过其方法.write()直接向可写流写入数据。

2. 使用模块stream中的函数pipeline()将可读流传输到可写流。

为了演示这些方法，我们使用它们来实现相同的函数writeIterableToFile()。

可读流的.pipe()方法也支持管道传输，但它有一个缺点，最好避免使用它。

9.5.2.1 writable.write(chunk)

在向流中写入数据时，有两种基于回调的机制可以帮助我们：

• 事件'drain'表示背压已经解除。

• 函数finished()在流：

  • 不再可读或可写

  • 已经遇到错误或过早关闭事件

在下一个示例中，我们将这些机制转换为 Promise，以便我们可以通过异步函数使用它们：

import * as util from 'util';
import * as stream from 'stream';
import * as fs from 'fs';
import {once} from 'events';

const finished = util.promisify(stream.finished); // (A)

async function writeIterableToFile(iterable, filePath) {
 const writable = fs.createWriteStream(filePath, {encoding: 'utf8'});
 for await (const chunk of iterable) {
 if (!writable.write(chunk)) { // (B)
 // Handle backpressure
 await once(writable, 'drain');
 }
 }
 writable.end(); // (C)
 // Wait until done. Throws if there are errors.
 await finished(writable);
}

await writeIterableToFile(
 ['One', ' line of text.\n'], 'tmp/log.txt');
assert.equal(
 fs.readFileSync('tmp/log.txt', {encoding: 'utf8'}),
 'One line of text.\n');

stream.finished()的默认版本是基于回调的，但可以通过util.promisify()（A 行）转换为基于 Promise 的版本。

我们使用了以下两种模式：

• 在处理背压的情况下向可写流写入数据（B 行）：

  if (!writable.write(chunk)) {
   await once(writable, 'drain');
  }
  

• 关闭可写流并等待写入完成（C 行）：

  writable.end();
  await finished(writable);
  

9.5.2.2 通过stream.pipeline()将可读流传输到可写流

在 A 行，我们使用stream.pipeline()的 Promise 版本将可读流readable传输到可写流writable：

import * as stream from 'stream';
import * as fs from 'fs';
const pipeline = util.promisify(stream.pipeline);

async function writeIterableToFile(iterable, filePath) {
 const readable = stream.Readable.from(
 iterable, {encoding: 'utf8'});
 const writable = fs.createWriteStream(filePath);
 await pipeline(readable, writable); // (A)
}
await writeIterableToFile(
 ['One', ' line of text.\n'], 'tmp/log.txt');
// ···

9.5.2.3 不推荐：readable.pipe(destination)

可读的.pipe()方法也支持管道传输，但有一个警告：如果可读流发出错误，则可写流不会自动关闭。pipeline()没有这个警告。

9.6 快速参考：与流相关的功能

模块os：

• const EOL: string（自 0.7.8 起）(https://nodejs.org/api/os.html#os_os_eol)

  包含当前平台使用的行尾字符序列。

模块buffer：

• Buffer.isEncoding(encoding: string): boolean（自 0.9.1 起）(https://nodejs.org/api/buffer.html#buffer_class_method_buffer_isencoding_encoding)

  如果encoding正确命名了受支持的 Node.js 文本编码之一，则返回true。支持的编码包括：

  • 'utf8'

  • 'utf16le'

  • 'ascii'

  • 'latin1

  • 'base64'

  • 'hex'（每个字节表示为两个十六进制字符）

模块stream：

• Readable.prototype[Symbol.asyncIterator](): AsyncIterableIterator<any>（自 10.0.0 起）(https://nodejs.org/api/stream.html#stream_readable_symbol_asynciterator)

  可读流是异步可迭代的。例如，您可以在异步函数或异步生成器中使用for-await-of循环来迭代它们。

• finished(stream: ReadableStream | WritableStream | ReadWriteStream, callback: (err?: ErrnoException | null) => void): () => Promise<void> (自 10.0.0 起)

  当读取/写入完成或出现错误时，返回的 Promise 将被解决。

  此 promisified 版本的创建方式如下：

  const finished = util.promisify(stream.finished);
  

• pipeline(...streams: Array<ReadableStream|ReadWriteStream|WritableStream>): Promise<void> (自 10.0.0 起)

  流之间的管道。当管道完成或出现错误时，返回的 Promise 将被解决。

  此 promisified 版本的创建方式如下：

  const pipeline = util.promisify(stream.pipeline);
  

• Readable.from(iterable: Iterable<any> | AsyncIterable<any>, options?: ReadableOptions): Readable (自 12.3.0 起)

  将可迭代对象转换为可读流。

  interface ReadableOptions {
   highWaterMark?: number;
   encoding?: string;
   objectMode?: boolean;
   read?(this: Readable, size: number): void;
   destroy?(this: Readable, error: Error | null,
   callback: (error: Error | null) => void): void;
   autoDestroy?: boolean;
  }
  

  这些选项与Readable构造函数的选项相同，并在此处有文档记录。

模块fs：

• createReadStream(path: string | Buffer | URL, options?: string | {encoding?: string; start?: number}): ReadStream (自 2.3.0 起)

  创建可读流。还有更多选项可用。

• createWriteStream(path: PathLike, options?: string | {encoding?: string; flags?: string; mode?: number; start?: number}): WriteStream (自 2.3.0 起)

  使用.flags选项，您可以指定是要写入还是追加，以及文件存在或不存在时会发生什么。还有更多选项可用。

本节中的静态类型信息基于Definitely Typed。

9.7 进一步阅读和本章的来源

• Node.js 文档中的“流与异步生成器和异步迭代器兼容性”章节

• “JavaScript for impatient programmers”中的“异步函数”章节

• “JavaScript for impatient programmers”中的“异步迭代”章节

评论

十、在 Node.js 上使用 web 流

原文：exploringjs.com/nodejs-shell-scripting/ch_web-streams.html

译者：飞龙

协议：CC BY-NC-SA 4.0

• 10.1 什么是 web 流？

  • 10.1.1 流的种类

  • 10.1.2 管道链

  • 10.1.3 背压

  • 10.1.4 Node.js 中对 web 流的支持

• 10.2 从 ReadableStreams 读取

  • 10.2.1 通过 Readers 消费 ReadableStreams

  • 10.2.2 通过异步迭代消费 ReadableStreams

  • 10.2.3 将 ReadableStreams 管道到 WritableStreams

• 10.3 通过包装将数据源转换为 ReadableStreams

  • 10.3.1 实现底层源的第一个示例

  • 10.3.2 使用 ReadableStream 包装推送源或拉取源

• 10.4 写入 WritableStreams

  • 10.4.1 通过 Writers 写入 WritableStreams

  • 10.4.2 管道到 WritableStreams

• 10.5 通过包装将数据汇转换为 WritableStreams

  • 10.5.1 示例：跟踪 ReadableStream

  • 10.5.2 示例：收集写入字符串的 WriteStream 块

• 10.6 使用 TransformStreams

  • 10.6.1 标准 TransformStreams

• 10.7 实现自定义 TransformStreams

  • 10.7.1 示例：将任意块的流转换为行流

  • 10.7.2 提示：异步生成器也非常适合转换流

• 10.8 更深入地了解背压

  • 10.8.1 信号背压

  • 10.8.2 对背压的反应

• 10.9 字节流

  • 10.9.1 可读字节流

  • 10.9.2 示例：填充随机数据的无限可读字节流

  • 10.9.3 示例：压缩可读字节流

  • 10.9.4 示例：通过 fetch() 读取网页

• 10.10 Node.js 特定的辅助函数

• 10.11 进一步阅读

Web 流 是一种标准的 流，现在在所有主要的 web 平台上都得到支持：web 浏览器、Node.js 和 Deno。（流是一种从各种来源顺序读取和写入数据的抽象，例如文件、托管在服务器上的数据等。）

例如，全局函数 fetch()（用于下载在线资源）异步返回一个具有 web 流属性 .body 的 Response。

本章涵盖了 Node.js 上的 web 流，但我们所学的大部分内容都适用于支持它们的所有 web 平台。

10.1 什么是网络流？

让我们首先概述一下网络流的一些基本知识。之后，我们将快速转移到示例。

流是一种用于访问数据的数据结构，例如：

• 文件

• 托管在 Web 服务器上的数据

• 等等。

它们的两个好处是：

• 我们可以处理大量数据，因为流允许我们将它们分割成较小的片段（所谓的chunks），我们可以一次处理一个。

• 我们可以在处理不同数据时使用相同的数据结构，流。这样可以更容易地重用代码。

Web streams（“web”通常被省略）是一个相对较新的标准，起源于 Web 浏览器，但现在也受到 Node.js 和 Deno 的支持（如此MDN 兼容性表所示）。

在网络流中，chunks 通常是：

• 文本流：字符串

• 二进制流：Uint8Arrays（一种 TypedArray）

10.1.1 流的种类

有三种主要类型的网络流：

• 一个 ReadableStream 用于从source读取数据。执行此操作的代码称为consumer。

• 一个 WritableStream 用于向sink写入数据。执行此操作的代码称为producer。

• TransformStream 由两个流组成：

  • 它从其writable side接收输入，即 WritableStream。

  • 它将输出发送到其readable side，即 ReadableStream。

  这个想法是通过“管道传输”TransformStream 来转换数据。也就是说，我们将数据写入可写端，并从可读端读取转换后的数据。以下 TransformStreams 内置在大多数 JavaScript 平台中（稍后会详细介绍）：

  • 因为 JavaScript 字符串是 UTF-16 编码的，所以在 JavaScript 中，UTF-8 编码的数据被视为二进制数据。TextDecoderStream将这样的数据转换为字符串。

  • TextEncoderStream将 JavaScript 字符串转换为 UTF-8 数据。

  • CompressionStream将二进制数据压缩为 GZIP 和其他压缩格式。

  • DecompressionStream从 GZIP 和其他压缩格式中解压缩二进制数据。

ReadableStreams，WritableStreams 和 TransformStreams 可用于传输文本或二进制数据。在本章中，我们将主要进行前者。 字节流用于二进制数据，在最后简要提到。

10.1.2 管道链路

Piping是一种操作，它让我们将一个 ReadableStream 连接到一个 WritableStream：只要 ReadableStream 产生数据，此操作就会读取该数据并将其写入 WritableStream。如果我们连接了两个流，我们就可以方便地将数据从一个位置传输到另一个位置（例如复制文件）。但是，我们也可以连接多于两个流，并获得可以以各种方式处理数据的管道链路。这是一个管道链路的例子：

• 它以一个 ReadableStream 开始。

• 接下来是一个或多个 TransformStreams。

• 链路以 WritableStream 结束。

通过将前者连接到后者的可写端，将一个 ReadableStream 连接到 TransformStream。类似地，通过将前者的可读端连接到后者的可写端，将一个 TransformStream 连接到另一个 TransformStream。并且通过将前者的可读端连接到后者的可写端，将一个 TransformStream 连接到一个 WritableStream。

10.1.3 背压

管道链路中的一个问题是，成员可能会收到比它目前能处理的更多数据。 背压是解决这个问题的一种技术：它使数据的接收者能够告诉发送者应该暂时停止发送数据，以便接收者不会被压倒。

另一种看待背压的方式是作为一个信号，通过管道链路向后传播，从被压倒的成员到链路的开始。例如，考虑以下管道链路：

ReadableStream -pipeTo-> TransformStream -pipeTo-> WriteableStream

这是背压通过这个链路传播的方式：

• 最初，WriteableStream 发出信号，表明它暂时无法处理更多数据。

• 管道停止从 TransformStream 中读取。

• 输入在 TransformStream 中积累（被缓冲）。

• TransformStream 发出满的信号。

• 管道停止从 ReadableStream 中读取。

我们已经到达管道链的开头。因此，在 ReadableStream 中没有数据积累（也被缓冲），WritableStream 有时间恢复。一旦它恢复，它会发出信号表明它已准备好再次接收数据。该信号也会通过链返回，直到它到达 ReadableStream，数据处理恢复。

在这第一次对背压的探讨中，为了让事情更容易理解，省略了一些细节。这些将在以后进行讨论。

10.1.4 Node.js 中对 web 流的支持

在 Node.js 中，Web 流可以从两个来源获得：

• 来自模块'node:stream/web'

• 通过全局变量（就像在 Web 浏览器中）

目前，只有一个 API 在 Node.js 中直接支持 web 流 – Fetch API：

const response = await fetch('https://example.com');
const readableStream = response.body;

对于其他事情，我们需要使用模块'node:stream'中以下静态方法之一，将 Node.js 流转换为 Web 流，反之亦然：

• Node.js 的 Readable 可以转换为 WritableStreams，反之亦然：

  • Readable.toWeb(nodeReadable)

  • Readable.fromWeb(webReadableStream, options?)

• Node.js 的 Writable 可以转换为 ReadableStreams，反之亦然：

  • Writable.toWeb(nodeWritable)

  • Writable.fromWeb(webWritableStream, options?)

• Node.js 的 Duplex 可以转换为 TransformStreams，反之亦然：

  • Duplex.toWeb(nodeDuplex)

  • Duplex.fromWeb(webTransformStream, options?)

还有一个 API 部分支持 web 流：FileHandles 有方法.readableWebStream()。

10.2 从 ReadableStreams 中读取

ReadableStreams 让我们从各种来源读取数据块。它们具有以下类型（随意浏览此类型及其属性的解释；当我们在示例中遇到它们时，它们将再次被解释）：

interface ReadableStream<TChunk> {
 getReader(): ReadableStreamDefaultReader<TChunk>;
 readonly locked: boolean;
 [Symbol.asyncIterator](): AsyncIterator<TChunk>;

 cancel(reason?: any): Promise<void>;

 pipeTo(
 destination: WritableStream<TChunk>,
 options?: StreamPipeOptions
 ): Promise<void>;
 pipeThrough<TChunk2>(
 transform: ReadableWritablePair<TChunk2, TChunk>,
 options?: StreamPipeOptions
 ): ReadableStream<TChunk2>;

 // Not used in this chapter:
 tee(): [ReadableStream<TChunk>, ReadableStream<TChunk>];
}

interface StreamPipeOptions {
 signal?: AbortSignal;
 preventClose?: boolean;
 preventAbort?: boolean;
 preventCancel?: boolean;
}

这些属性的解释：

• .getReader()返回一个 Reader – 通过它我们可以从 ReadableStream 中读取。ReadableStreams 返回 Readers 类似于可迭代对象返回迭代器。

• .locked: 一次只能有一个活动的 Reader 读取 ReadableStream。当一个 Reader 正在使用时，ReadableStream 被锁定，无法调用.getReader()。

• [Symbol.asyncIterator](https://exploringjs.com/impatient-js/ch_async-iteration.html): 这个方法使得 ReadableStreams 可以异步迭代。目前只在一些平台上实现。

• .cancel(reason)取消流，因为消费者对它不再感兴趣。reason被传递给 ReadableStream 的底层源的.cancel()方法（稍后会详细介绍）。返回的 Promise 在此操作完成时实现。

• .pipeTo()将其 ReadableStream 的内容传送到 WritableStream。返回的 Promise 在此操作完成时实现。.pipeTo()确保背压、关闭、错误等都正确地通过管道链传播。我们可以通过它的第二个参数指定选项：

  • .signal让我们向这个方法传递一个 AbortSignal，这使我们能够通过 AbortController 中止管道传输。

  • .preventClose: 如果为true，它会阻止在 ReadableStream 关闭时关闭 WritableStream。当我们想要将多个 ReadableStream 管道到同一个 WritableStream 时，这是有用的。

  • 其余选项超出了本章的范围。它们在web 流规范中有文档记录。

• .pipeThrough()将其 ReadableStream 连接到一个 ReadableWritablePair（大致是一个 TransformStream，稍后会详细介绍）。它返回生成的 ReadableStream（即 ReadableWritablePair 的可读端）。

以下小节涵盖了三种消费 ReadableStreams 的方式：

• 通过 Readers 进行读取

• 通过异步迭代进行读取

• 将 ReadableStreams 连接到 WritableStreams

10.2.1 通过 Reader 消费 ReadableStreams

我们可以使用Readers从 ReadableStreams 中读取数据。它们具有以下类型（随意浏览此类型及其属性的解释；当我们在示例中遇到它们时，它们将再次被解释）：

interface ReadableStreamGenericReader {
 readonly closed: Promise<undefined>;
 cancel(reason?: any): Promise<void>;
}
interface ReadableStreamDefaultReader<TChunk>
 extends ReadableStreamGenericReader
{
 releaseLock(): void;
 read(): Promise<ReadableStreamReadResult<TChunk>>;
}

interface ReadableStreamReadResult<TChunk> {
 done: boolean;
 value: TChunk | undefined;
}

这些属性的解释：

• .closed：此 Promise 在流关闭后被满足。如果流出现错误或者在流关闭之前 Reader 的锁被释放，它将被拒绝。

• .cancel()：在活动的 Reader 中，此方法取消关联的 ReadableStream。

• .releaseLock()停用 Reader 并解锁其流。

• .read()返回一个 Promise，用于 ReadableStreamReadResult（一个包装的块），它有两个属性：

  • .done是一个布尔值，只要可以读取块，就为false，在最后一个块之后为true。

  • .value是块（或在最后一个块之后是undefined）。

如果您了解迭代的工作原理，ReadableStreamReadResult 可能会很熟悉：ReadableStreams 类似于可迭代对象，Readers 类似于迭代器，而 ReadableStreamReadResults 类似于迭代器方法.next()返回的对象。

以下代码演示了使用 Readers 的协议：

const reader = readableStream.getReader(); // (A)
assert.equal(readableStream.locked, true); // (B)
try {
 while (true) {
 const {done, value: chunk} = await reader.read(); // (C)
 if (done) break;
 // Use `chunk`
 }
} finally {
 reader.releaseLock(); // (D)
}

**获取 Reader。**我们不能直接从readableStream中读取，我们首先需要获取一个Reader（行 A）。每个 ReadableStream 最多可以有一个 Reader。获取 Reader 后，readableStream被锁定（行 B）。在我们可以再次调用.getReader()之前，我们必须调用.releaseLock()（行 D）。

读取块。.read()返回一个带有属性.done和.value的对象的 Promise（行 C）。在读取最后一个块之后，.done为true。这种方法类似于 JavaScript 中异步迭代的工作方式。

10.2.1.1 示例：通过 ReadableStream 读取文件

在下面的示例中，我们从文本文件data.txt中读取块（字符串）：

import * as fs from 'node:fs';
import {Readable} from 'node:stream';

const nodeReadable = fs.createReadStream(
 'data.txt', {encoding: 'utf-8'});
const webReadableStream = Readable.toWeb(nodeReadable); // (A)

const reader = webReadableStream.getReader();
try {
 while (true) {
 const {done, value} = await reader.read();
 if (done) break;
 console.log(value);
 }
} finally {
 reader.releaseLock();
}
// Output:
// 'Content of text file\n'

我们将 Node.js Readable 转换为 web ReadableStream（行 A）。然后我们使用先前解释的协议来读取块。

10.2.1.2 示例：使用 ReadableStream 内容组装字符串

在下一个示例中，我们将所有 ReadableStream 的块连接成一个字符串并返回它：

/**
 * Returns a string with the contents of `readableStream`.
 */
async function readableStreamToString(readableStream) {
 const reader = readableStream.getReader();
 try {
 let result = '';
 while (true) {
 const {done, value} = await reader.read();
 if (done) {
 return result; // (A)
 }
 result += value;
 }
 } finally {
 reader.releaseLock(); // (B)
 }
}

方便的是，finally子句总是被执行 - 无论我们如何离开try子句。也就是说，如果我们返回一个结果（行 A），锁将被正确释放（行 B）。

10.2.2 通过异步迭代消费 ReadableStreams

ReadableStreams 也可以通过异步迭代进行消费：

const iterator = readableStream[Symbol.asyncIterator]();
let exhaustive = false;
try {
 while (true) {
 let chunk;
 ({done: exhaustive, value: chunk} = await iterator.next());
 if (exhaustive) break;
 console.log(chunk);
 }
} finally {
 // If the loop was terminated before we could iterate exhaustively
 // (via an exception or `return`), we must call `iterator.return()`.
 // Check if that was the case.
 if (!exhaustive) {
 iterator.return();
 }
}

值得庆幸的是，for-await-of循环为我们处理了异步迭代的所有细节：

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

10.2.2.1 示例：使用异步迭代读取流

让我们重新尝试从文件中读取文本。这次，我们使用异步迭代而不是 Reader：

import * as fs from 'node:fs';
import {Readable} from 'node:stream';

const nodeReadable = fs.createReadStream(
 'text-file.txt', {encoding: 'utf-8'});
const webReadableStream = Readable.toWeb(nodeReadable);
for await (const chunk of webReadableStream) {
 console.log(chunk);
}
// Output:
// 'Content of text file'

10.2.2.2 示例：使用 ReadableStream 内容组装字符串

我们以前使用 Reader 来组装一个包含 ReadableStream 内容的字符串。有了异步迭代，代码变得更简单了：

/**
 * Returns a string with the contents of `readableStream`.
 */
async function readableStreamToString2(readableStream) {
 let result = '';
 for await (const chunk of readableStream) {
 result += chunk;
 }
 return result;
}

10.2.2.3 注意事项：浏览器不支持对 ReadableStreams 进行异步迭代

目前，Node.js 和 Deno 支持对 ReadableStreams 进行异步迭代，但 Web 浏览器不支持：有一个 GitHub 问题链接到错误报告。

鉴于尚不完全清楚浏览器将如何支持异步迭代，包装比填充更安全。以下代码基于Chromium bug 报告中的建议：

async function* getAsyncIterableFor(readableStream) {
 const reader = readableStream.getReader();
 try {
 while (true) {
 const {done, value} = await reader.read();
 if (done) return;
 yield value;
 }
 } finally {
 reader.releaseLock();
 }
}

10.2.3 将可读流导入可写流

可读流有两种管道方法：

• readableStream.pipeTo(writeableStream)同步返回一个 Promise p。它异步读取readableStream的所有块，并将它们写入writableStream。完成后，它会实现p。

  当我们探索可写流时，我们将看到.pipeTo()的示例，因为它提供了一种方便的方式将数据传输到其中。

• readableStream.pipeThrough(transformStream)将readableStream导入transformStream.writable并返回transformStream.readable（每个 TransformStream 都有这些属性，它们指向其可写侧和可读侧）。另一种看待这个操作的方式是，我们通过连接transformStream到readableStream创建一个新的可读流。

  当我们探索 TransformStreams 时，我们将看到.pipeThrough()的示例，因为这是它们主要使用的方法。

10.3 将数据源通过包装转换为可读流

如果我们想通过一个可读流读取外部源，我们可以将其包装在一个适配器对象中，并将该对象传递给ReadableStream构造函数。适配器对象被称为可读流的底层源（当我们更仔细地看 backpressure 时，将解释排队策略）：

new ReadableStream(underlyingSource?, queuingStrategy?)

这是底层源的类型（随意浏览此类型及其属性的解释；当我们在示例中遇到它们时，它们将再次解释）：

interface UnderlyingSource<TChunk> {
 start?(
 controller: ReadableStreamController<TChunk>
 ): void | Promise<void>;
 pull?(
 controller: ReadableStreamController<TChunk>
 ): void | Promise<void>;
 cancel?(reason?: any): void | Promise<void>;

 // Only used in byte streams and ignored in this section:
 type: 'bytes' | undefined;
 autoAllocateChunkSize: bigint;
}

这是当可读流调用这些方法时：

• 在调用ReadableStream的构造函数后立即调用.start(controller)。

• 每当可读流的内部队列中有空间时，都会调用.pull(controller)。直到队列再次满了为止，它会被重复调用。此方法只会在.start()完成后调用。如果.pull()没有入队任何内容，它将不会再次被调用。

• 如果可读流的消费者通过readableStream.cancel()或reader.cancel()取消它，将调用.cancel(reason)。reason是传递给这些方法的值。

这些方法中的每一个都可以返回一个 Promise，并且在 Promise 解决之前不会采取进一步的步骤。如果我们想要做一些异步操作，这是有用的。

.start()和.pull()的参数controller让它们访问流。它具有以下类型：

type ReadableStreamController<TChunk> =
 | ReadableStreamDefaultController<TChunk>
 | ReadableByteStreamController<TChunk> // ignored here
;

interface ReadableStreamDefaultController<TChunk> {
 enqueue(chunk?: TChunk): void;
 readonly desiredSize: number | null;
 close(): void;
 error(err?: any): void;
}

现在，块是字符串。我们稍后将介绍字节流，其中 Uint8Arrays 很常见。这些方法的作用是：

• .enqueue(chunk)将chunk添加到可读流的内部队列。

• .desiredSize指示.enqueue()写入的队列中有多少空间。如果队列已满，则为零，如果超过了最大大小，则为负。因此，如果期望大小为零或负，则我们必须停止入队。

  • 如果流关闭，其期望大小为零。

  • 如果流处于错误模式，其期望大小为null。

• .close()关闭可读流。消费者仍然可以清空队列，但之后，流将结束。底层源调用此方法很重要-否则，读取其流将永远不会结束。

• .error(err)将流置于错误模式：以后与它的所有交互都将以错误值err失败。

10.3.1 实现底层源的第一个示例

在我们实现底层源的第一个示例中，我们只提供了.start()方法。我们将在下一小节中看到.pull()的用例。

const readableStream = new ReadableStream({
 start(controller) {
 controller.enqueue('First line\n'); // (A)
 controller.enqueue('Second line\n'); // (B)
 controller.close(); // (C)
 },
});
for await (const chunk of readableStream) {
 console.log(chunk);
}

// Output:
// 'First line\n'
// 'Second line\n'

我们使用控制器创建一个具有两个块（行 A 和行 B）的流。关闭流很重要（行 C）。否则，for-await-of循环永远不会结束！

请注意，这种入队的方式并不完全安全：存在超出内部队列容量的风险。我们很快将看到如何避免这种风险。

10.3.2 使用 ReadableStream 包装推送源或拉取源

一个常见的场景是将推送源或拉取源转换为 ReadableStream。源是推送还是拉取决定了我们将如何与 UnderlyingSource 连接到 ReadableStream：

• 推送源：这样的源在有新数据时通知我们。我们使用.start()来设置监听器和支持数据结构。如果我们收到太多数据，期望的大小不再是正数，我们必须告诉我们的源暂停。如果以后调用了.pull()，我们可以取消暂停。对外部源在期望的大小变为非正数时暂停的反应称为应用背压。

• 拉取源：我们向这样的源请求新数据-通常是异步的。因此，我们通常在.start()中不做太多事情，并在调用.pull()时检索数据。

接下来我们将看到两种来源的例子。

10.3.2.1 示例：从具有背压支持的推送源创建一个 ReadableStream

在下面的示例中，我们将一个 ReadableStream 包装在一个套接字周围-它向我们推送数据（它调用我们）。这个例子来自 web 流规范：

function makeReadableBackpressureSocketStream(host, port) {
 const socket = createBackpressureSocket(host, port);

 return new ReadableStream({
 start(controller) {
 socket.ondata = event => {
 controller.enqueue(event.data);

 if (controller.desiredSize <= 0) {
 // The internal queue is full, so propagate
 // the backpressure signal to the underlying source.
 socket.readStop();
 }
 };

 socket.onend = () => controller.close();
 socket.onerror = () => controller.error(
 new Error('The socket errored!'));
 },

 pull() {
 // This is called if the internal queue has been emptied, but the
 // stream’s consumer still wants more data. In that case, restart
 // the flow of data if we have previously paused it.
 socket.readStart();
 },

 cancel() {
 socket.close();
 },
 });
}

10.3.2.2 示例：从拉取源创建一个 ReadableStream

工具函数iterableToReadableStream()接受一个块的可迭代对象，并将其转换为一个 ReadableStream：

/**
 * @param  iterable an iterable (asynchronous or synchronous)
 */
 function iterableToReadableStream(iterable) {
 return new ReadableStream({
 start() {
 if (typeof iterable[Symbol.asyncIterator] === 'function') {
 this.iterator = iterable[Symbol.asyncIterator]();
 } else if (typeof iterable[Symbol.iterator] === 'function') {
 this.iterator = iterable[Symbol.iterator]();
 } else {
 throw new Error('Not an iterable: ' + iterable);
 }
 },

 async pull(controller) {
 if (this.iterator === null) return;
 // Sync iterators return non-Promise values,
 // but `await` doesn’t mind and simply passes them on
 const {value, done} = await this.iterator.next();
 if (done) {
 this.iterator = null;
 controller.close();
 return;
 }
 controller.enqueue(value);
 },

 cancel() {
 this.iterator = null;
 controller.close();
 },
 });
}

让我们使用一个异步生成器函数来创建一个异步可迭代对象，并将该可迭代对象转换为一个 ReadableStream：

async function* genAsyncIterable() {
 yield 'how';
 yield 'are';
 yield 'you';
}
const readableStream = iterableToReadableStream(genAsyncIterable());
for await (const chunk of readableStream) {
 console.log(chunk);
}

// Output:
// 'how'
// 'are'
// 'you'

iterableToReadableStream()也适用于同步可迭代对象：

const syncIterable = ['hello', 'everyone'];
const readableStream = iterableToReadableStream(syncIterable);
for await (const chunk of readableStream) {
 console.log(chunk);
}

// Output:
// 'hello'
// 'everyone'

可能会有一个静态的辅助方法ReadableStream.from()，提供这个功能（请参阅其拉取请求以获取更多信息）。

10.4 向 WritableStreams 写入

WritableStreams 让我们向各种接收器写入数据块。它们具有以下类型（随意浏览此类型和其属性的解释；当我们在示例中遇到它们时，它们将再次解释）：

interface WritableStream<TChunk> {
 getWriter(): WritableStreamDefaultWriter<TChunk>;
 readonly locked: boolean;

 close(): Promise<void>;
 abort(reason?: any): Promise<void>;
}

这些属性的解释：

• .getWriter()返回一个 Writer-通过它我们可以向 WritableStream 写入数据的对象。

• .locked：WritableStream 一次只能有一个活动的 Writer。当一个 Writer 正在使用时，WritableStream 被锁定，无法调用.getWriter()。

• .close()关闭流：

  • 底层接收器（稍后会详细介绍）在关闭之前仍将接收所有排队的块。

  • 从现在开始，所有的写入尝试都将无声地失败（没有错误）。

  • 该方法返回一个 Promise，如果接收器成功写入所有排队的块并关闭，将实现该 Promise。如果在这些步骤中发生任何错误，它将被拒绝。

• .abort()中止流：

  • 它将流置于错误模式。

  • 返回的 Promise 在接收器成功关闭时实现，如果发生错误则拒绝。

以下小节涵盖了向 WritableStreams 发送数据的两种方法：

• 通过 Writers 向 WritableStreams 写入

• 将数据传输到 WritableStreams

10.4.1 通过 Writers 向 WritableStreams 写入

我们可以使用Writers向 WritableStreams 写入。它们具有以下类型（随意浏览此类型和其属性的解释；当我们在示例中遇到它们时，它们将再次解释）：

interface WritableStreamDefaultWriter<TChunk> {
 readonly desiredSize: number | null;
 readonly ready: Promise<undefined>;
 write(chunk?: TChunk): Promise<void>;
 releaseLock(): void;

 close(): Promise<void>;
 readonly closed: Promise<undefined>;
 abort(reason?: any): Promise<void>;
}

这些属性的解释：

• .desiredSize指示 WriteStream 队列中有多少空间。如果队列已满，则为零，如果超过最大大小，则为负数。因此，如果期望的大小为零或负数，我们必须停止写入。

  • 如果流关闭，它的期望大小为零。

  • 如果流处于错误模式，它的期望大小为null。

• .ready返回一个 Promise，在期望的大小从非正数变为正数时实现。这意味着没有背压活动，可以写入数据。如果期望的大小后来再次变为非正数，则会创建并返回一个新的待处理 Promise。

• .write()将一个块写入流。它返回一个 Promise，在写入成功后实现，如果有错误则拒绝。

• .releaseLock()释放 Writer 对其流的锁定。

• .close()具有与关闭 Writer 流相同的效果。

• .closed返回一个 Promise，在流关闭时被实现。

• .abort()具有与中止 Writer 流相同的效果。

以下代码显示了使用 Writers 的协议：

const writer = writableStream.getWriter(); // (A)
assert.equal(writableStream.locked, true); // (B)
try {
 // Writing the chunks (explained later)
} finally {
 writer.releaseLock(); // (C)
}

我们不能直接向writableStream写入，我们首先需要获取一个Writer（A 行）。每个 WritableStream 最多只能有一个 Writer。在获取了 Writer 之后，writableStream被锁定（B 行）。在我们可以再次调用.getWriter()之前，我们必须调用.releaseLock()（C 行）。

有三种写入块的方法。

10.4.1.1 写入方法 1：等待.write()（处理背压效率低下）

第一种写入方法是等待每个.write()的结果：

await writer.write('Chunk 1');
await writer.write('Chunk 2');
await writer.close();

由.write()返回的 Promise 在我们传递给它的块成功写入时实现。“成功写入”具体意味着什么取决于 WritableStream 的实现方式 - 例如，对于文件流，该块可能已发送到操作系统，但仍然驻留在缓存中，因此实际上尚未写入磁盘。

由.close()返回的 Promise 在流关闭时实现。

这种写入方法的一个缺点是等待写入成功意味着队列没有被使用。因此，数据吞吐量可能会较低。

10.4.1.2 写入方法 2：忽略.write()拒绝（忽略背压）

在第二种写入方法中，我们忽略了.write()返回的 Promise，只等待.close()返回的 Promise：

writer.write('Chunk 1').catch(() => {}); // (A)
writer.write('Chunk 2').catch(() => {}); // (B)
await writer.close(); // reports errors

.write()的同步调用将块添加到 WritableStream 的内部队列中。通过不等待返回的 Promises，我们不必等待每个块被写入。但是，等待.close()确保队列为空，并且所有写入都成功后我们才继续。

在 A 行和 B 行调用.catch()是必要的，以避免在写入过程中出现问题时出现有关未处理的 Promise 拒绝的警告。这样的警告通常会记录在控制台上。我们可以忽略.write()报告的错误，因为.close()也会向我们报告这些错误。

通过使用一个忽略 Promise 拒绝的辅助函数，可以改进先前的代码：

ignoreRejections(
 writer.write('Chunk 1'),
 writer.write('Chunk 2'),
);
await writer.close(); // reports errors

function ignoreRejections(...promises) {
 for (const promise of promises) {
 promise.catch(() => {});
 }
}

这种方法的一个缺点是忽略了背压：我们只是假设队列足够大，可以容纳我们写入的所有内容。

10.4.1.3 写入方法 3：等待.ready（高效处理背压）

在这种写入方法中，我们通过等待 Writer getter.ready来有效地处理背压：

await writer.ready; // reports errors
// How much room do we have?
console.log(writer.desiredSize);
writer.write('Chunk 1').catch(() => {});

await writer.ready; // reports errors
// How much room do we have?
console.log(writer.desiredSize);
writer.write('Chunk 2').catch(() => {});

await writer.close(); // reports errors

.ready中的 Promise 在流从有背压到无背压的转换时实现。

10.4.1.4 示例：通过 Writer 写入文件

在这个例子中，我们通过 WritableStream 创建一个文本文件data.txt：

import * as fs from 'node:fs';
import {Writable} from 'node:stream';

const nodeWritable = fs.createWriteStream(
 'new-file.txt', {encoding: 'utf-8'}); // (A)
const webWritableStream = Writable.toWeb(nodeWritable); // (B)

const writer = webWritableStream.getWriter();
try {
 await writer.write('First line\n');
 await writer.write('Second line\n');
 await writer.close();
} finally {
 writer.releaseLock()
}

在 A 行，我们为文件data.txt创建了一个 Node.js 流。在 B 行，我们将这个流转换为 web 流。然后我们使用 Writer 将字符串写入其中。

10.4.2 向 WritableStreams 进行管道传输

除了使用 Writers，我们还可以通过将 ReadableStreams 传输到 WritableStreams 来向 WritableStreams 写入：

await readableStream.pipeTo(writableStream);

由.pipeTo()返回的 Promise 在传输成功完成时实现。

10.4.2.1 管道传输是异步进行的

管道传输是在当前任务完成或暂停后执行的。以下代码演示了这一点：

const readableStream = new ReadableStream({ // (A)
 start(controller) {
 controller.enqueue('First line\n');
 controller.enqueue('Second line\n');
 controller.close();
 },
});
const writableStream = new WritableStream({ // (B)
 write(chunk) {
 console.log('WRITE: ' + JSON.stringify(chunk));
 },
 close() {
 console.log('CLOSE WritableStream');
 },
});

console.log('Before .pipeTo()');
const promise = readableStream.pipeTo(writableStream); // (C)
promise.then(() => console.log('Promise fulfilled'));
console.log('After .pipeTo()');

// Output:
// 'Before .pipeTo()'
// 'After .pipeTo()'
// 'WRITE: "First line\n"'
// 'WRITE: "Second line\n"'
// 'CLOSE WritableStream'
// 'Promise fulfilled'

在 A 行我们创建一个 ReadableStream。在 B 行我们创建一个 WritableStream。

我们可以看到.pipeTo()（行 C）立即返回。在一个新的任务中，块被读取和写入。然后writableStream被关闭，最后，promise被实现。

10.4.2.2 示例：将数据管道到文件的可写流

在下面的示例中，我们为一个文件创建一个 WritableStream，并将一个 ReadableStream 管道传递给它：

const webReadableStream = new ReadableStream({ // (A)
 async start(controller) {
 controller.enqueue('First line\n');
 controller.enqueue('Second line\n');
 controller.close();
 },
});

const nodeWritable = fs.createWriteStream( // (B)
 'data.txt', {encoding: 'utf-8'});
const webWritableStream = Writable.toWeb(nodeWritable); // (C)

await webReadableStream.pipeTo(webWritableStream); // (D)

在 A 行，我们创建了一个 ReadableStream。在 B 行，我们为文件data.txt创建了一个 Node.js 流。在 C 行，我们将这个流转换为 web 流。在 D 行，我们将我们的webReadableStream管道传递给文件的 WritableStream。

10.4.2.3 示例：将两个 ReadableStreams 写入到一个 WritableStream

在下面的示例中，我们将两个 ReadableStreams 写入单个 WritableStream。

function createReadableStream(prefix) {
 return new ReadableStream({
 async start(controller) {
 controller.enqueue(prefix + 'chunk 1');
 controller.enqueue(prefix + 'chunk 2');
 controller.close();
 },
 });
}

const writableStream = new WritableStream({
 write(chunk) {
 console.log('WRITE ' + JSON.stringify(chunk));
 },
 close() {
 console.log('CLOSE');
 },
 abort(err) {
 console.log('ABORT ' + err);
 },
});

await createReadableStream('Stream 1: ')
 .pipeTo(writableStream, {preventClose: true}); // (A)
await createReadableStream('Stream 2: ')
 .pipeTo(writableStream, {preventClose: true}); // (B)
await writableStream.close();

// Output
// 'WRITE "Stream 1: chunk 1"'
// 'WRITE "Stream 1: chunk 2"'
// 'WRITE "Stream 2: chunk 1"'
// 'WRITE "Stream 2: chunk 2"'
// 'CLOSE'

我们告诉.pipeTo()在 ReadableStream 关闭后不关闭 WritableStream（行 A 和行 B）。因此，在行 A 之后，WritableStream 保持打开状态，我们可以将另一个 ReadableStream 管道传递给它。

10.5 将数据接收端通过包装转换为可写流

如果我们想通过 WritableStream 写入到外部接收端，我们可以将其包装在一个适配器对象中，并将该对象传递给WritableStream的构造函数。适配器对象被称为 WritableStream 的底层接收端（当我们更仔细地看反压时，排队策略将在稍后解释）：

new WritableStream(underlyingSink?, queuingStrategy?)

这是底层接收端的类型（随意浏览此类型和其属性的解释；当我们在示例中遇到它们时，它们将再次解释）：

interface UnderlyingSink<TChunk> {
 start?(
 controller: WritableStreamDefaultController
 ): void | Promise<void>;
 write?(
 chunk: TChunk,
 controller: WritableStreamDefaultController
 ): void | Promise<void>;
 close?(): void | Promise<void>;;
 abort?(reason?: any): void | Promise<void>;
}

这些属性的解释：

• .start(controller) 在我们调用WritableStream的构造函数后立即调用。如果我们做一些异步操作，我们可以返回一个 Promise。在这个方法中，我们可以准备写入。

• .write(chunk, controller) 当一个新的块准备写入外部接收端时调用。我们可以通过返回一个 Promise 来施加反压，一旦反压消失就会实现。

• .close() 在调用writer.close()后调用，并且所有排队的写入都成功。在这个方法中，我们可以在写入后进行清理。

• 如果调用了writeStream.abort()或writer.abort()，则会调用.abort(reason)。reason是传递给这些方法的值。

.start()和.write()的参数controller让它们错误 WritableStream。它具有以下类型：

interface WritableStreamDefaultController {
 readonly signal: AbortSignal;
 error(err?: any): void;
}

• .signal 是一个 AbortSignal，如果我们想在流被中止时中止写入或关闭操作，我们可以监听它。

• .error(err) 错误 WritableStream：它被关闭，并且以后所有与它的交互都会失败，错误值为err。

10.5.1 示例：跟踪一个可读流

在下一个示例中，我们将一个 ReadableStream 管道到一个 WritableStream，以便检查 ReadableStream 如何生成块：

const readableStream = new ReadableStream({
 start(controller) {
 controller.enqueue('First chunk');
 controller.enqueue('Second chunk');
 controller.close();
 },
});
await readableStream.pipeTo(
 new WritableStream({
 write(chunk) {
 console.log('WRITE ' + JSON.stringify(chunk));
 },
 close() {
 console.log('CLOSE');
 },
 abort(err) {
 console.log('ABORT ' + err);
 },
 })
);
// Output:
// 'WRITE "First chunk"'
// 'WRITE "Second chunk"'
// 'CLOSE'

10.5.2 示例：收集写入到 WriteStream 的块到一个字符串中

在下一个示例中，我们创建了WriteStream的一个子类，它将所有写入的块收集到一个字符串中。我们可以通过.getString()方法访问该字符串：

class StringWritableStream extends WritableStream {
 #string = '';
 constructor() {
 super({
 // We need to access the `this` of `StringWritableStream`.
 // Hence the arrow function (and not a method).
 write: (chunk) => {
 this.#string += chunk;
 },
 });
 }
 getString() {
 return this.#string;
 }
}
const stringStream = new StringWritableStream();
const writer = stringStream.getWriter();
try {
 await writer.write('How are');
 await writer.write(' you?');
 await writer.close();
} finally {
 writer.releaseLock()
}
assert.equal(
 stringStream.getString(),
 'How are you?'
);

这种方法的一个缺点是我们混合了两个 API：WritableStream的 API 和我们新的字符串流 API。另一种选择是委托给 WritableStream 而不是扩展它：

function StringcreateWritableStream() {
 let string = '';
 return {
 stream: new WritableStream({
 write(chunk) {
 string += chunk;
 },
 }),
 getString() {
 return string;
 },
 };
}

const stringStream = StringcreateWritableStream();
const writer = stringStream.stream.getWriter();
try {
 await writer.write('How are');
 await writer.write(' you?');
 await writer.close();
} finally {
 writer.releaseLock()
}
assert.equal(
 stringStream.getString(),
 'How are you?'
);

这个功能也可以通过类来实现（而不是作为对象的工厂函数）。

10.6 使用 TransformStreams

一个 TransformStream：

• 通过其writable side接收输入，即 WritableStream。

• 然后可能会或可能不会转换这个输入。

• 结果可以通过一个 ReadableStream 来读取，它的可读端。

使用 TransformStreams 最常见的方式是“管道传递”它们：

const transformedStream = readableStream.pipeThrough(transformStream);

.pipeThrough() 将readableStream管道到transformStream的可写端，并返回其可读端。换句话说：我们已经创建了一个新的ReadableStream，它是readableStream的转换版本。

.pipeThrough() 不仅接受 TransformStreams，还接受任何具有以下形式的对象：

interface ReadableWritablePair<RChunk, WChunk> {
 readable: ReadableStream<RChunk>;
 writable: WritableStream<WChunk>;
}

10.6.1 标准 TransformStreams

Node.js 支持以下标准 TransformStreams：

• 编码（WHATWG 标准） – TextEncoderStream和TextDecoderStream：

  • 这些流支持 UTF-8，但也支持许多“旧编码”。

  • 一个 Unicode 代码点被编码为多达四个 UTF-8 代码单元（字节）。在字节流中，编码的代码点可能会跨越块。TextDecoderStream可以正确处理这些情况。

  • 大多数 JavaScript 平台都可以使用（TextEncoderStream, TextDecoderStream）。

• 压缩流（W3C 草案社区组报告） – CompressionStream, DecompressionStream:

  • 当前支持的压缩格式: deflate（ZLIB 压缩数据格式），deflate-raw（DEFLATE 算法），gzip（GZIP 文件格式）。

  • 在许多 JavaScript 平台上都可以使用（CompressionStream, DecompressionStream）。

10.6.1.1 示例：解码一系列 UTF-8 编码的字节流

在下面的示例中，我们解码了一系列 UTF-8 编码的字节流：

const response = await fetch('https://example.com');
const readableByteStream = response.body;
const readableStream = readableByteStream
 .pipeThrough(new TextDecoderStream('utf-8'));
for await (const stringChunk of readableStream) {
 console.log(stringChunk);
}

response.body是一个 ReadableByteStream，其块是Uint8Array的实例（TypedArrays）。我们通过TextDecoderStream将该流传输，以获得具有字符串块的流。

请注意，单独翻译每个字节块（例如通过TextDecoder）是行不通的，因为一个 Unicode 代码点在 UTF-8 中被编码为多达四个字节，而这些字节可能不都在同一个块中。

10.6.1.2 示例：创建一个用于标准输入的可读文本流

以下 Node.js 模块记录通过标准输入发送给它的所有内容：

// echo-stdin.mjs
import {Readable} from 'node:stream';

const webStream = Readable.toWeb(process.stdin)
 .pipeThrough(new TextDecoderStream('utf-8'));
for await (const chunk of webStream) {
 console.log('>>>', chunk);
}

我们可以通过存储在process.stdin中的流访问标准输入（process是一个全局 Node.js 变量）。如果我们不为此流设置编码并通过Readable.toWeb()进行转换，我们将获得一个字节流。我们通过 TextDecoderStream 将其传输，以获得一个文本流。

请注意，我们逐步处理标准输入：一旦另一个块可用，我们就会记录它。换句话说，我们不会等到标准输入完成。当数据要么很大要么只是间歇性发送时，这是很有用的。

10.7 实现自定义 TransformStreams

我们可以通过将 Transformer 对象传递给TransformStream的构造函数来实现自定义 TransformStream。这样的对象具有以下类型（随意略过此类型和其属性的解释；当我们在示例中遇到它们时，它们将再次解释）：

interface Transformer<TInChunk, TOutChunk> {
 start?(
 controller: TransformStreamDefaultController<TOutChunk>
 ): void | Promise<void>;
 transform?(
 chunk: TInChunk,
 controller: TransformStreamDefaultController<TOutChunk>
 ): void | Promise<void>;
 flush?(
 controller: TransformStreamDefaultController<TOutChunk>
 ): void | Promise<void>;
}

这些属性的解释：

• .start(controller)在我们调用TransformStream的构造函数之后立即调用。在这里，我们可以在转换开始之前准备好一些东西。

• .transform(chunk, controller)执行实际的转换。它接收一个输入块，并可以使用其参数controller来排队一个或多个转换后的输出块。它也可以选择不排队任何内容。

• .flush(controller)在所有输入块成功转换后调用。在这里，我们可以在转换完成后执行清理工作。

这些方法中的每一个都可以返回一个 Promise，并且在 Promise 解决之前不会采取进一步的步骤。如果我们想要执行一些异步操作，这是很有用的。

参数controller具有以下类型：

interface TransformStreamDefaultController<TOutChunk> {
 enqueue(chunk?: TOutChunk): void;
 readonly desiredSize: number | null;
 terminate(): void;
 error(err?: any): void;
}

• .enqueue(chunk)将chunk添加到 TransformStream 的可读端（输出）。

• .desiredSize返回可读端（输出）的 TransformStream 内部队列的期望大小。

• .terminate()关闭可读端（输出）并错误可写端（输入）的 TransformStream。如果转换器对可写端（输入）的剩余块不感兴趣并希望跳过它们，则可以使用它。

• .error(err)错误 TransformStream：以后所有与它的交互都将以错误值err失败。

TransformStream 中的背压如何？该类将背压从其可读端（输出）传播到其可写端（输入）。假设转换不会改变数据量太多。因此，Transform 可以忽略背压。但是，可以通过transformStreamDefaultController.desiredSize检测到它，并通过从transformer.transform()返回一个 Promise 来传播它。

10.7.1 示例：将任意块的流转换为行流

TransformStream的以下子类将流转换为每个块都包含一行文本的流。也就是说，除了最后一个块可能以行尾（EOL）字符串结束之外，每个块都以行尾（EOL）字符串结束：Unix（包括 macOS）上为'\n'，Windows 上为'\r\n'。

class ChunksToLinesTransformer {
 #previous = '';

 transform(chunk, controller) {
 let startSearch = this.#previous.length;
 this.#previous += chunk;
 while (true) {
 // Works for EOL === '\n' and EOL === '\r\n'
 const eolIndex = this.#previous.indexOf('\n', startSearch);
 if (eolIndex < 0) break;
 // Line includes the EOL
 const line = this.#previous.slice(0, eolIndex+1);
 controller.enqueue(line);
 this.#previous = this.#previous.slice(eolIndex+1);
 startSearch = 0;
 }
 }

 flush(controller) {
 // Clean up and enqueue any text we’re still holding on to
 if (this.#previous.length > 0) {
 controller.enqueue(this.#previous);
 }
 }
}
class ChunksToLinesStream extends TransformStream {
 constructor() {
 super(new ChunksToLinesTransformer());
 }
}

const stream = new ReadableStream({
 async start(controller) {
 controller.enqueue('multiple\nlines of\ntext');
 controller.close();
 },
});
const transformStream = new ChunksToLinesStream();
const transformed = stream.pipeThrough(transformStream);

for await (const line of transformed) {
 console.log('>>>', JSON.stringify(line));
}

// Output:
// '>>> "multiple\n"'
// '>>> "lines of\n"'
// '>>> "text"'

请注意，Deno 的内置TextLineStream提供类似的功能。

提示：我们也可以通过异步生成器进行这种转换。它将异步迭代 ReadableStream 并返回一个包含行的异步可迭代对象。其实现在§9.4“通过异步生成器转换可读流”中显示。

10.7.2 提示：异步生成器也非常适合转换流

由于 ReadableStreams 是异步可迭代的，我们可以使用异步生成器来转换它们。这导致非常优雅的代码：

const stream = new ReadableStream({
 async start(controller) {
 controller.enqueue('one');
 controller.enqueue('two');
 controller.enqueue('three');
 controller.close();
 },
});

async function* prefixChunks(prefix, asyncIterable) {
 for await (const chunk of asyncIterable) {
 yield '> ' + chunk;
 }
}

const transformedAsyncIterable = prefixChunks('> ', stream);
for await (const transformedChunk of transformedAsyncIterable) {
 console.log(transformedChunk);
}

// Output:
// '> one'
// '> two'
// '> three'

10.8 仔细观察背压

让我们仔细观察背压。考虑以下管道链：

rs.pipeThrough(ts).pipeTo(ws);

rs是一个 ReadableStream，ts是一个 TransformStream，ws是一个 WritableStream。这些是由前一个表达式创建的连接（.pipeThrough使用.pipeTo将rs连接到ts的可写端）：

rs -pipeTo-> ts{writable,readable} -pipeTo-> ws

观察：

• rs的基础源可以被视为在rs之前的管道链成员。

• ws的基础接收器可以被视为在ws之后的管道链成员。

• 每个流都有一个内部缓冲区：ReadableStreams 在其基础源之后进行缓冲。WritableStreams 在其基础接收器之前进行缓冲。

假设ws的基础接收器速度慢，ws的缓冲区最终满了。然后发生以下步骤：

• ws发出满的信号。

• pipeTo停止从ts.readable读取。

• ts.readable发出满的信号。

• ts停止从ts.writable移动块到ts.readable。

• ts.writable发出满的信号。

• pipeTo停止从rs读取。

• rs向其基础源发出满的信号。

• 基础源暂停。

这个例子说明我们需要两种功能：

• 接收数据的实体需要能够发出背压信号。

• 发送数据的实体需要对信号做出反应，施加背压。

让我们探索这些功能在 web 流 API 中是如何实现的。

10.8.1 发出背压

背压由接收数据的实体发出信号。Web 流有两个这样的实体：

• WritableStream 通过 Writer 方法.write()接收数据。

• 当其基础源调用 ReadableStreamDefaultController 方法.enqueue()时，ReadableStream 接收数据。

在这两种情况下，输入都通过队列进行缓冲。施加背压的信号是队列已满。让我们看看如何检测到这一点。

这些是队列的位置：

• 一个 WritableStream 的队列在 WritableStreamDefaultController 中内部存储（参见 web 流标准）。

• 一个 ReadableStream 的队列在 ReadableStreamDefaultController 中内部存储（参见 web 流标准）。

队列的期望大小是一个数字，表示队列中还有多少空间：

• 如果队列中仍有空间，则为正。

• 如果队列已达到其最大大小，则为零。

• 如果队列已超过其最大大小，则为负。

因此，如果期望的大小为零或更少，我们必须施加背压。它可以通过包含队列的对象的 getter.desiredSize获得。

期望的大小是如何计算的？通过指定所谓的排队策略的对象。ReadableStream和WritableStream具有默认的排队策略，可以通过它们的构造函数的可选参数进行覆盖。接口QueuingStrategy有两个属性：

• 方法.size(chunk)返回chunk的大小。

  • 队列的当前大小是它包含的块的大小之和。

• 属性.highWaterMark指定队列的最大大小。

队列的期望大小是高水位标记减去队列的当前大小。

10.8.2 对背压的反应

发送数据的实体需要对信号背压做出反应，通过施加背压。

10.8.2.1 通过 Writer 写入 WritableStream 的代码

• 我们可以在writer.ready中等待 Promise。在等待期间，我们被阻塞，期望的背压得到了实现。一旦队列中有空间，Promise 就会被实现。当writer.desiredSize的值大于零时，实现会被触发。

• 或者，我们可以等待writer.write()返回的 Promise。如果我们这样做，队列甚至不会被填满。

如果我们愿意，我们还可以根据writer.desiredSize来确定我们的块的大小。

10.8.2.2 ReadableStream 的底层源

可以传递给 ReadableStream 的底层源对象包装了外部源。在某种程度上，它也是管道链的成员；在其 ReadableStream 之前的成员。

• 只有在队列中有空间时，才会要求底层拉取源提供新数据。在没有空间时，会自动施加背压，因为没有数据被拉取。

• 在入队后，底层推送源应检查controller.desiredSize：如果为零或更少，则应通过暂停其外部源来施加背压。

10.8.2.3 WritableStream 的底层接收端

可以传递给 WritableStream 的底层接收端对象包装了外部接收端。在某种程度上，它也是管道链的成员；在其 WritableStream 之后的成员。

每个外部接收端以不同的方式（在某些情况下根本不）信号背压。底层接收端可以通过从方法.write()返回一个被实现的 Promise 来施加背压，一旦写入完成。在web 流标准中有一个例子，演示了这是如何工作的。

10.8.2.4 一个 transformStream（.writable → .readable）

TransformStream 通过为前者实现底层接收端和为后者实现底层源，将其可写端连接到其可读端。它具有一个内部插槽.[[backpressure]]，指示内部背压当前是否处于活动状态。

• 可写端的底层接收器的.write()方法会异步等待，直到没有内部背压，然后将另一个块提供给 TransformStream 的转换器（web streams 标准：TransformStreamDefaultSinkWriteAlgorithm）。然后转换器可以通过其 TransformStreamDefaultController 加入一些内容。请注意，.write()返回一个 Promise，在方法完成时会被满足。在此之前，WriteStream 通过其队列缓冲传入的写请求。因此，可写端的背压通过该队列及其期望的大小来表示。

• 如果通过 TransformStreamDefaultController 将一个块加入队列，并且可读端的队列变满了，TransformStream 的背压就会被激活（web streams 标准：TransformStreamDefaultControllerEnqueue）。

• 如果从读取器中读取了一些内容，ReadableStream的背压可能会被取消（web streams 标准：ReadableStreamDefaultReaderRead）：

  • 如果队列中现在有空间，可能是时候调用底层源的.pull()了（web streams 标准：.[[PullSteps]]）。

  • 可读端的底层源的.pull()会取消背压（web streams 标准：TransformStreamDefaultSourcePullAlgorithm）。

10.8.2.5 .pipeTo()（ReadableStream → WritableStream）

.pipeTo()通过读取器从 ReadableStream 读取块，并通过写入器将它们写入 WritableStream。当writer.desiredSize为零或更小时，它会暂停（web streams 标准：ReadableStreamPipeTo的第 15 步）。

10.9 字节流

到目前为止，我们只使用过文本流，流的块是字符串。但是 web streams API 也支持字节流，用于二进制数据，其中块是 Uint8Arrays（TypedArrays）：

• ReadableStream有一个特殊的'bytes'模式。

• WritableStream本身不关心块是字符串还是 Uint8Arrays。因此，实例是文本流还是字节流取决于底层接收器可以处理什么类型的块。

• TransformStream可以处理什么类型的块也取决于其 Transformer。

接下来，我们将学习如何创建可读的字节流。

10.9.1 可读的字节流

ReadableStream构造函数创建的流的类型取决于可选的属性.type和可选的第一个参数underlyingSource：

• 如果.type被省略或没有提供底层源，则新实例是一个文本流。

• 如果.type是字符串'bytes'，则新实例是一个字节流：

  const readableByteStream = new ReadableStream({
   type: 'bytes',
   async start() { /*...*/ }
   // ...
  });
  

如果一个 ReadableStream 处于'bytes'模式，会发生什么变化？

在默认模式下，底层源可以返回任何类型的块。在字节模式下，块必须是 ArrayBufferViews，即 TypedArrays（例如 Uint8Arrays）或 DataViews。

此外，可读的字节流可以创建两种读取器：

• .getReader()返回一个ReadableStreamDefaultReader的实例。

• .getReader({mode: 'byob'}) 返回一个 ReadableStreamBYOBReader 的实例。

“BYOB” 代表 “Bring Your Own Buffer”，意味着我们可以传递一个缓冲区（ArrayBufferView）给 reader.read()。之后，该 ArrayBufferView 将被分离并且不再可用。但是.read() 返回其数据在一个新的 ArrayBufferView 中，该 ArrayBufferView 具有相同的类型并访问相同的 ArrayBuffer 的相同区域。

此外，可读的字节流具有不同的控制器：它们是ReadableByteStreamController的实例（而不是ReadableStreamDefaultController）。除了强制底层源将 ArrayBufferViews（TypedArrays 或 DataViews）入队之外，它还通过其属性.byobRequest支持 ReadableStreamBYOBReaders。底层源将其数据写入存储在此属性中的 BYOBRequest。Web 流标准在其“创建流的示例”部分中有两个使用.byobRequest的示例。

10.9.2 示例：填充随机数据的无限可读的字节流

在下一个示例中，创建一个无限可读的字节流，用随机数据填充其块（灵感来自：example4.mjs in “在 Node.js 中实现 Web 流 API”）。

import {promisify} from 'node:util';
import {randomFill} from 'node:crypto';
const asyncRandomFill = promisify(randomFill);

const readableByteStream = new ReadableStream({
 type: 'bytes',
 async pull(controller) {
 const byobRequest = controller.byobRequest;
 await asyncRandomFill(byobRequest.view);
 byobRequest.respond(byobRequest.view.byteLength);
 },
});

const reader = readableByteStream.getReader({mode: 'byob'});
const buffer = new Uint8Array(10); // (A)
const firstChunk = await reader.read(buffer); // (B)
console.log(firstChunk);

由于readableByteStream是无限的，我们无法循环读取它。这就是为什么我们只读取它的第一个块（B 行）。

我们在 A 行创建的缓冲区在 B 行之后被传输，因此无法读取。

10.9.3 示例：压缩可读的字节流

在下面的示例中，我们创建一个可读的字节流，并将其通过一个将其压缩为 GZIP 格式的流：

const readableByteStream = new ReadableStream({
 type: 'bytes',
 start(controller) {
 // 256 zeros
 controller.enqueue(new Uint8Array(256));
 controller.close();
 },
});
const transformedStream = readableByteStream.pipeThrough(
 new CompressionStream('gzip'));
await logChunks(transformedStream);

async function logChunks(readableByteStream) {
 const reader = readableByteStream.getReader();
 try {
 while (true) {
 const {done, value} = await reader.read();
 if (done) break;
 console.log(value);
 }
 } finally {
 reader.releaseLock();
 }
}

10.9.4 示例：通过fetch()读取网页

fetch()的结果解析为一个响应对象，其属性.body是一个可读的字节流。我们通过TextDecoderStream将该字节流转换为文本流：

const response = await fetch('https://example.com');
const readableByteStream = response.body;
const readableStream = readableByteStream.pipeThrough(
 new TextDecoderStream('utf-8'));
for await (const stringChunk of readableStream) {
 console.log(stringChunk);
}

10.10 Node.js 特定的辅助函数

Node.js 是唯一支持以下辅助函数的 Web 平台，它称之为实用消费者：

import {
 arrayBuffer,
 blob,
 buffer,
 json,
 text,
} from 'node:stream/consumers';

这些函数将 Web ReadableStreams、Node.js Readables 和 AsyncIterators 转换为被满足的 Promise：

• ArrayBuffers（arrayBuffer()）

• Blobs（blob()）

• Node.js 缓冲区（buffer()）

• JSON 对象（json()）

• 字符串（text()）

假定二进制数据为 UTF-8 编码：

import * as streamConsumers from 'node:stream/consumers';

const readableByteStream = new ReadableStream({
 type: 'bytes',
 start(controller) {
 // TextEncoder converts strings to UTF-8 encoded Uint8Arrays
 const encoder = new TextEncoder();
 const view = encoder.encode('"😀"');
 assert.deepEqual(
 view,
 Uint8Array.of(34, 240, 159, 152, 128, 34)
 );
 controller.enqueue(view);
 controller.close();
 },
});
const jsonData = await streamConsumers.json(readableByteStream);
assert.equal(jsonData, '😀');

字符串流按预期工作：

import * as streamConsumers from 'node:stream/consumers';

const readableByteStream = new ReadableStream({
 start(controller) {
 controller.enqueue('"😀"');
 controller.close();
 },
});
const jsonData = await streamConsumers.json(readableByteStream);
assert.equal(jsonData, '😀');

10.11 进一步阅读

本节提到的所有材料都是本章的来源。

本章不涵盖 Web 流 API 的每个方面。您可以在此处找到更多信息：

• “WHATWG 流标准” by Adam Rice, Domenic Denicola, Mattias Buelens, and 吉野剛史 (Takeshi Yoshino)

• “Web Streams API” in Node.js 文档

更多材料：

• Web 流 API：

  • “在 Node.js 中实现 Web 流 API” by James M. Snell

  • “流 API” 在 MDN 上

  • “流-权威指南” by Thomas Steiner

• 背压：

  • “Node.js 流中的背压” by Vladimir Topolev

  • “流中的背压” in Node.js 文档

• Unicode（代码点，UTF-8，UTF-16 等）：“Unicode 简介”章节 in “JavaScript for impatient programmers”

• “异步迭代”章节 in “JavaScript for impatient programmers”

• “Typed Arrays：处理二进制数据”章节 in “JavaScript for impatient programmers”

评论

十一、流配方

exploringjs.com/nodejs-shell-scripting/ch_stream-recipes.html

• 11.1 写入标准输出(stdout)

  • 11.1.1 通过console.log()写入 stdout

  • 11.1.2 通过 Node.js 流写入 stdout

  • 11.1.3 通过 Web 流写入 stdout

• 11.2 写入标准错误(stderr)

• 11.3 从标准输入(stdin)读取

  • 11.3.1 通过 Node.js 流从 stdin 读取

  • 11.3.2 通过 Web 流从 stdin 读取

  • 11.3.3 通过模块'node:readline'从 stdin 读取

• 11.4 Node.js 流配方

• 11.5 Web 流配方

11.1 写入标准输出(stdout)

这是写入 stdout 的三个选项：

• 我们可以通过console.log()写入它。

• 我们可以通过 Node.js 流写入它。

• 我们可以通过 Web 流写入它。

11.1.1 通过console.log()写入 stdout

console.log(format, ...args)写入 stdout 并始终附加换行符'\n'（即使在 Windows 上也是如此）。第一个参数可以包含占位符，这些占位符的解释方式与util.format()相同：

console.log('String: %s Number: %d Percent: %%', 'abc', 123);

const obj = {one: 1, two: 2};
console.log('JSON: %j Object: %o', obj, obj);

// Output:
// 'String: abc Number: 123 Percent: %'
// 'JSON: {"one":1,"two":2} Object: { one: 1, two: 2 }'

第一个参数之后的所有参数始终显示在输出中，即使没有足够的占位符。

11.1.2 通过 Node.js 流写入 stdout

process.stdout是stream.Readable的一个实例。这意味着我们可以像使用其他 Node.js 流一样使用它-例如：

process.stdout.write('two');
process.stdout.write(' words');
process.stdout.write('\n');

前面的代码等同于：

console.log('two words');

请注意，这种情况下末尾没有换行符，因为console.log()总是会添加一个。

如果我们使用.write()来处理大量数据，我们应该考虑回压，如§9.5.2.1“writable.write(chunk)”中所解释的那样。

以下配方适用于process.stdout：§11.4“Node.js 流配方”。

11.1.3 通过 Web 流写入 stdout

我们可以将process.stdout转换为 Web 流并写入其中：

import {Writable} from 'node:stream';
const webOut = Writable.toWeb(process.stdout);
const writer = webOut.getWriter();
try {
 await writer.write('First line\n');
 await writer.write('Second line\n');
 await writer.close();
} finally {
 writer.releaseLock()
}

以下配方适用于webOut：§11.5“Web 流配方”。

11.2 写入标准错误(stderr)

写入 stderr 的工作方式与写入 stdout 类似：

• 我们可以通过console.error()写入它。

• 我们可以通过 Node.js 流写入它。

• 我们可以通过 Web 流写入它。

有关更多信息，请参阅前一节。

11.3 从标准输入(stdin)读取

这些是从 stdin 读取的选项：

• 我们可以通过 Node.js 流从中读取。

• 我们可以通过 Web 流从中读取。

• 我们可以使用模块'node:readline'。

11.3.1 通过 Node.js 流从 stdin 读取

process.stdin是stream.Writable的一个实例。这意味着我们可以像使用其他 Node.js 流一样使用它：

// Switch to text mode (otherwise we get chunks of binary data)
process.stdin.setEncoding('utf-8');
for await (const chunk of process.stdin) {
 console.log('>', chunk);
}

以下配方适用于webIn：§11.4“Node.js 流配方”。

11.3.2 通过 Web 流从 stdin 读取

我们首先必须将process.stdin转换为 Web 流：

import {Readable} from 'node:stream';
// Switch to text mode (otherwise we get chunks of binary data)
process.stdin.setEncoding('utf-8');
const webIn = Readable.toWeb(process.stdin);
for await (const chunk of webIn) {
 console.log('>', chunk);
}

以下配方适用于webIn：§11.5“Web 流配方”。

11.3.3 通过模块'node:readline'从 stdin 读取

内置模块'node:readline'允许我们提示用户以交互方式输入信息-例如：

import * as fs from 'node:fs';
import * as readline from 'node:readline/promises';

const rl = readline.createInterface({
 input: process.stdin,
 output: process.stdout,
});

const filePath = await rl.question('Please enter a file path: ');
fs.writeFileSync(filePath, 'Hi!', {encoding: 'utf-8'})

rl.close();

有关模块'node:readline'的更多信息，请参见：

• §9.3.3“通过模块’node:readlines’从可读流中读取行”

• 官方文档。

11.4 Node.js 流配方

可读流：

• §9.3.1.2“Readable.from(): 从可迭代对象创建可读流”

• §9.3.2“通过for-await-of从可读流中读取块”

  • §9.3.2.1“在字符串中收集可读流的内容”

• §9.3.3“通过模块’node:readlines’从可读流中读取行”

• §9.4“通过异步生成器转换可读流”

  • §9.4.1“在异步可迭代对象中从块转换为编号行”

可写流：

• §9.5.2“写入可写流”

• §9.5.2.2“通过stream.pipeline()将可读流传输到可写流”

11.5 网络流配方

从中创建一个 ReadableStream：

• 字符串：§10.3.1“实现基础源的第一个示例”

• 可迭代对象：§10.3.2.2“示例：从拉取源创建一个 ReadableStream”

从 ReadableStream 中读取：

• §10.2.1“通过读取器消耗 ReadableStreams”

• §10.2.2“通过异步迭代消耗 ReadableStreams”

  • §10.2.2.2“示例：组装包含 ReadableStream 内容的字符串”

• §10.2.3“将 ReadableStreams 传输到 WritableStreams”

转换 ReadableStreams：

• §10.6“使用 TransformStreams”

• §10.7.2“提示：异步生成器也非常适合转换流”

• §10.7.1“示例：将任意块的流转换为行流”

使用 WritableStreams：

• §10.4“写入可写流”

• §10.5.2“示例：在字符串中收集写入到 WriteStream 的块”

评论

十二、在子进程中运行 shell 命令

原文：exploringjs.com/nodejs-shell-scripting/ch_nodejs-child-process.html

译者：飞龙

协议：CC BY-NC-SA 4.0

• 12.1 本章概述](ch_nodejs-child-process.html#overview-of-this-chapter)

  • 12.1.1 Windows vs. Unix

  • 12.1.2 我们在示例中经常使用的功能

• 12.2 异步生成进程：spawn()

  • 12.2.1 spawn()的工作原理

  • 12.2.2 何时执行 shell 命令？](ch_nodejs-child-process.html#when-is-the-shell-command-executed)

  • 12.2.3 仅命令模式 vs. 参数模式](ch_nodejs-child-process.html#spawn-argument-modes)

  • 12.2.4 向子进程的 stdin 发送数据

  • 12.2.5 手动进行管道传输](ch_nodejs-child-process.html#piping-manually)

  • 12.2.6 处理不成功的退出（包括错误）

  • 12.2.7 等待子进程退出](ch_nodejs-child-process.html#waiting-for-the-exit-of-a-child-process)

  • 12.2.8 终止子进程](ch_nodejs-child-process.html#terminating-child-processes)

• 12.3 同步生成进程：spawnSync()

  • 12.3.1 何时执行 shell 命令？](ch_nodejs-child-process.html#when-is-the-shell-command-executed-1)

  • 12.3.2 从 stdout 读取](ch_nodejs-child-process.html#reading-from-stdout)

  • 12.3.3 向子进程的 stdin 发送数据

  • 12.3.4 处理不成功的退出（包括错误）

• 12.4 基于spawn()的异步辅助函数

  • 12.4.1 exec()

  • 12.4.2 execFile()

• 12.5 基于spawnAsync()的同步辅助函数

  • 12.5.1 execSync()

  • 12.5.2 execFileSync()

• 12.6 有用的库](ch_nodejs-child-process.html#useful-libraries)

  • 12.6.1 tinysh：用于生成 shell 命令的辅助程序

  • 12.6.2 node-powershell：通过 Node.js 执行 Windows PowerShell 命令

• 12.7 在模块'node:child_process'的功能之间进行选择

在本章中，我们将探讨如何通过模块'node:child_process'从 Node.js 执行 shell 命令。

12.1 本章概述

模块'node:child_process'有一个用于执行 shell 命令（在生成的子进程中）的函数，有两个版本：

• 一个异步版本的spawn()。

• 一个同步版本的spawnSync()。

我们将首先探讨spawn()，然后是spawnSync()。最后，我们将看一下基于它们并且相对类似的以下函数：

• 基于spawn()：

  • exec()

  • execFile()

• 基于spawnSync()：

  • execSync()

  • execFileSync()

12.1.1 Windows vs. Unix

本章中显示的代码在 Unix 上运行，但我也在 Windows 上进行了测试-其中大部分代码需要进行轻微更改（例如以'\r\n'而不是'\n'结尾）。

12.1.2 我们在示例中经常使用的功能

以下功能在示例中经常出现。这就是为什么在这里解释一次：

• 断言：对于原始值使用assert.equal()，对于对象使用assert.deepEqual()。示例中从未显示必要的导入：

  import * as assert from 'node:assert/strict';
  

• 函数Readable.toWeb()将 Node 的原生stream.Readable转换为 web 流（ReadableStream的实例）。这在§10“在 Node.js 上使用 web 流”中有解释。示例中始终导入Readable。

• 异步函数readableStreamToString()会消耗可读的 web 流并返回一个字符串（包装在 Promise 中）。这在 web 流章节中有解释。假定这个函数在示例中是可用的。

12.2 异步生成进程：spawn()

12.2.1 spawn()的工作原理

spawn(
 command: string,
 args?: Array<string>,
 options?: Object
): ChildProcess

spawn() 异步地在新进程中执行命令：该进程与 Node 的主 JavaScript 进程并行运行，我们可以通过各种方式与其通信（通常通过流）。

接下来，有关spawn()的参数和结果的文档。如果您喜欢通过示例学习，可以跳过该内容，继续阅读后面的小节。

12.2.1.1 参数：command

command是一个包含 shell 命令的字符串。有两种使用该参数的模式：

• 仅命令模式：省略args，command包含整个 shell 命令。我们甚至可以使用 shell 功能，如在多个可执行文件之间进行管道传输，将 I/O 重定向到文件，变量和通配符。

  • options.shell必须为true，因为我们需要一个 shell 来处理 shell 功能。

• 参数模式：command仅包含命令的名称，args包含其参数。

  • 如果options.shell为true，则参数中的许多元字符会被解释，并且通配符和变量名称等功能会起作用。

  • 如果options.shell为false，则字符串会直接使用，我们不必转义元字符。

这两种模式在本章后面进行了演示。

12.2.1.2 参数：options

以下options最有趣：

• .shell: boolean|string（默认值：false）

  是否应使用 shell 来执行命令？

  • 在 Windows 上，此选项几乎总是应为true。例如，否则无法执行.bat和.cmd文件。

  • 在 Unix 上，只有核心 shell 功能（例如管道，I/O 重定向，文件名通配符和变量）在.shell为false时不可用。

  • 如果.shell为true，我们必须小心处理用户输入并对其进行清理，因为很容易执行任意代码。如果我们想将其用作非元字符，则还必须转义元字符。

  • 我们还可以将.shell设置为 shell 可执行文件的路径。然后 Node.js 将使用该可执行文件来执行命令。如果我们将.shell设置为true，Node.js 将使用：

    • Unix：'/bin/sh'

    • Windows：process.env.ComSpec

• .cwd: string | URL

  指定在执行命令时要使用的当前工作目录（CWD）。

• .stdio: Array<string|Stream>|string

  配置标准 I/O 的设置方式。下面会有解释。

• .env: Object（默认值：process.env）

  让我们为子进程指定 shell 变量。提示：

  • 查看process.env（例如在 Node.js REPL 中）以查看存在哪些变量。

  • 我们可以使用扩展运算符来非破坏性地覆盖现有变量 - 或者如果尚不存在，则创建它：

    {env: {...process.env, MY_VAR: 'Hi!'}}
    

• .signal: AbortSignal

  如果我们创建了一个 AbortController ac，我们可以将ac.signal传递给spawn()，并通过ac.abort()中止子进程。这在本章后面有演示。

• .timeout: number

  如果子进程的执行时间超过.timeout毫秒，则会被终止。

12.2.1.3 options.stdio

子进程的每个标准 I/O 流都有一个数字 ID，称为文件描述符：

• 标准输入（stdin）的文件描述符为 0。

• 标准输出(stdout)的文件描述符为 1。

• 标准错误(stderr)的文件描述符为 2。

可能会有更多的文件描述符，但这很少见。

options.stdio配置子进程的流是否以及如何被管道连接到父进程的流。它可以是一个数组，其中每个元素配置等于其索引的文件描述符。可以使用以下值作为数组元素：

• 'pipe'：

  • 索引 0：将childProcess.stdin管道连接到子进程的 stdin。请注意，尽管其名称如此，但前者是属于父进程的流。

  • 索引 1：将子进程的 stdout 管道连接到childProcess.stdout。

  • 索引 2：将子进程的 stderr 管道连接到childProcess.stderr。

• 'ignore'：忽略子进程的流。

• 'inherit'：将子进程的流管道连接到父进程的相应流。

  • 例如，如果我们希望子进程的 stderr 被记录到控制台，我们可以在索引 2 处使用'inherit'。

• 原生 Node.js 流：管道到该流或从该流。

• 还支持其他值，但这超出了本章的范围。

除了通过数组指定options.stdio之外，我们还可以缩写：

• 'pipe'等同于['pipe', 'pipe', 'pipe']（options.stdio的默认值）。

• 'ignore'等同于['ignore', 'ignore', 'ignore']。

• 'inherit'等同于['inherit', 'inherit', 'inherit']。

12.2.1.4 结果：ChildProcess的实例

spawn()返回ChildProcess的实例。

有趣的数据属性：

• .exitCode: number | null

  包含子进程退出时的代码：

  • 0（零）表示正常退出。

  • 大于零的数字表示发生了错误。

  • null表示进程尚未退出。

• .signalCode: string | null

  子进程被杀死的 POSIX 信号，或者如果没有被杀死则为null。有关更多信息，请参阅下面的.kill()方法的描述。

• 流：根据标准 I/O 的配置方式（请参阅前面的小节），以下流变得可用：

  • .stdin

  • .stdout

  • .stderr

• .pid: number | undefined

  子进程的进程标识符（PID）。如果生成失败，.pid为undefined。在调用spawn()后立即可用此值。

有趣的方法：

• .kill(signalCode?: number | string = 'SIGTERM'): boolean

  向子进程发送 POSIX 信号（通常导致进程终止）：

  • signal的 man 页面包含值的列表。

  • Windows 不支持信号，但 Node.js 模拟了其中一些 - 例如：SIGINT，SIGTERM和SIGKILL。有关更多信息，请参阅Node.js 文档。

  此方法在本章后面进行了演示。

有趣的事件：

• .on('exit', (exitCode: number|null, signalCode: string|null) => {})

  此事件在子进程结束后发出：

  • 回调参数为我们提供了退出代码或信号代码：其中一个始终为非空。

  • 由于多个进程可能共享相同的流，因此其标准 I/O 流可能仍然打开。事件'close'在子进程退出后通知我们所有 stdio 流都已关闭。

• .on('error', (err: Error) => {})

  如果进程无法被生成（请参阅示例后面）或子进程无法被杀死，则最常见地发出此事件。在此事件之后可能会或可能不会发出'exit'事件。

我们稍后将看到如何将事件转换为可以等待的 Promise。

12.2.2 shell 命令何时执行？

在使用异步spawn()时，命令的子进程是异步启动的。以下代码演示了这一点：

import {spawn} from 'node:child_process';

spawn(
 'echo', ['Command starts'],
 {
 stdio: 'inherit',
 shell: true,
 }
);
console.log('After spawn()');

这是输出：

After spawn()
Command starts

12.2.3 仅命令模式 vs. 参数模式

在本节中，我们以两种方式指定相同的命令调用：

• 仅命令模式：我们通过第一个参数command提供整个调用。

• 参数模式：我们通过第一个参数command提供命令，通过第二个参数args提供参数。

12.2.3.1 仅命令模式

import {Readable} from 'node:stream';
import {spawn} from 'node:child_process';

const childProcess = spawn(
 'echo "Hello, how are you?"',
 {
 shell: true, // (A)
 stdio: ['ignore', 'pipe', 'inherit'], // (B)
 }
);
const stdout = Readable.toWeb(
 childProcess.stdout.setEncoding('utf-8'));

// Result on Unix
assert.equal(
 await readableStreamToString(stdout),
 'Hello, how are you?\n' // (C)
);

// Result on Windows: '"Hello, how are you?"\r\n'

每个带参数的仅命令生成都需要.shell为true（A 行）-即使它像这个这么简单。

在 B 行，我们告诉spawn()如何处理标准 I/O：

• 忽略标准输入。

• 将子进程的标准输出管道到childProcess.stdout（属于父进程的流）。

• 将子进程的标准错误输出管道到父进程的标准错误输出。

在这种情况下，我们只对子进程的输出感兴趣。因此，一旦我们处理了输出，我们就完成了。在其他情况下，我们可能需要等到子进程退出。如何做到这一点，稍后会有演示。

在仅命令模式下，我们看到 shell 的更多特殊之处 - 例如，Windows 命令 shell 输出包括双引号（最后一行）。

12.2.3.2 参数模式

import {Readable} from 'node:stream';
import {spawn} from 'node:child_process';

const childProcess = spawn(
 'echo', ['Hello, how are you?'],
 {
 shell: true,
 stdio: ['ignore', 'pipe', 'inherit'],
 }
);
const stdout = Readable.toWeb(
 childProcess.stdout.setEncoding('utf-8'));

// Result on Unix
assert.equal(
 await readableStreamToString(stdout),
 'Hello, how are you?\n'
);
// Result on Windows: 'Hello, how are you?\r\n'

12.2.3.3 args中的元字符

让我们探讨一下如果args中有元字符会发生什么：

import {Readable} from 'node:stream';
import {spawn} from 'node:child_process';

async function echoUser({shell, args}) {
 const childProcess = spawn(
 `echo`, args,
 {
 stdio: ['ignore', 'pipe', 'inherit'],
 shell,
 }
 );
 const stdout = Readable.toWeb(
 childProcess.stdout.setEncoding('utf-8'));
 return readableStreamToString(stdout);
}

// Results on Unix
assert.equal(
 await echoUser({shell: false, args: ['$USER']}), // (A)
 '$USER\n'
);
assert.equal(
 await echoUser({shell: true, args: ['$USER']}), // (B)
 'rauschma\n'
);
assert.equal(
 await echoUser({shell: true, args: [String.raw`\$USER`]}), // (C)
 '$USER\n'
);

• 如果我们不使用 shell，例如美元符号（$）等元字符没有效果（A 行）。

• 在 shell 中，$USER被解释为一个变量（B 行）。

• 如果我们不想要这个，我们必须通过反斜杠转义美元符号（C 行）。

其他元字符（如星号（*））也会产生类似的效果。

这是 Unix shell 元字符的两个例子。Windows shell 有它们自己的元字符和它们自己的转义方式。

12.2.3.4 一个更复杂的 shell 命令

让我们使用更多的 shell 特性（这需要仅命令模式）：

import {Readable} from 'node:stream';
import {spawn} from 'node:child_process';
import {EOL} from 'node:os';

const childProcess = spawn(
 `(echo cherry && echo apple && echo banana) | sort`,
 {
 stdio: ['ignore', 'pipe', 'inherit'],
 shell: true,
 }
);
const stdout = Readable.toWeb(
 childProcess.stdout.setEncoding('utf-8'));
assert.equal(
 await readableStreamToString(stdout),
 'apple\nbanana\ncherry\n'
);

12.2.4 将数据发送到子进程的标准输入

到目前为止，我们只读取了子进程的标准输出。但是我们也可以将数据发送到标准输入：

import {Readable, Writable} from 'node:stream';
import {spawn} from 'node:child_process';

const childProcess = spawn(
 `sort`, // (A)
 {
 stdio: ['pipe', 'pipe', 'inherit'],
 }
);
const stdin = Writable.toWeb(childProcess.stdin); // (B)
const writer = stdin.getWriter(); // (C)
try {
 await writer.write('Cherry\n');
 await writer.write('Apple\n');
 await writer.write('Banana\n');
} finally {
 writer.close();
}

const stdout = Readable.toWeb(
 childProcess.stdout.setEncoding('utf-8'));
assert.equal(
 await readableStreamToString(stdout),
 'Apple\nBanana\nCherry\n'
);

我们使用 shell 命令sort（A 行）来为我们对文本行进行排序。

在 B 行，我们使用Writable.toWeb()将本机 Node.js 流转换为网络流（更多信息，请参见§10“在 Node.js 上使用网络流”）。

如何通过写入器（C 行）向 WritableStream 写入也在网络流章节中有解释。

12.2.5 手动进行管道传输

我们之前让 shell 执行以下命令：

(echo cherry && echo apple && echo banana) | sort

在下面的例子中，我们手动进行管道传输，从 echo（A 行）到 sorting（B 行）：

import {Readable, Writable} from 'node:stream';
import {spawn} from 'node:child_process';

const echo = spawn( // (A)
 `echo cherry && echo apple && echo banana`,
 {
 stdio: ['ignore', 'pipe', 'inherit'],
 shell: true,
 }
);
const sort = spawn( // (B)
 `sort`,
 {
 stdio: ['pipe', 'pipe', 'inherit'],
 shell: true,
 }
);

//==== Transferring chunks from echo.stdout to sort.stdin ====

const echoOut = Readable.toWeb(
 echo.stdout.setEncoding('utf-8'));
const sortIn = Writable.toWeb(sort.stdin);

const sortInWriter = sortIn.getWriter();
try {
 for await (const chunk of echoOut) { // (C)
 await sortInWriter.write(chunk);
 }
} finally {
 sortInWriter.close();
}

//==== Reading sort.stdout ====

const sortOut = Readable.toWeb(
 sort.stdout.setEncoding('utf-8'));
assert.equal(
 await readableStreamToString(sortOut),
 'apple\nbanana\ncherry\n'
);

例如echoOut这样的 ReadableStreams 是异步可迭代的。这就是为什么我们可以使用for-await-of循环来读取它们的chunks（流数据的片段）。更多信息，请参见§10“在 Node.js 上使用网络流”。

12.2.6 处理不成功的退出（包括错误）

有三种主要的不成功的退出方式：

• 子进程无法生成。

• Shell 中发生了错误。

• 一个进程被终止。

12.2.6.1 子进程无法生成

以下代码演示了如果子进程无法生成会发生什么。在这种情况下，原因是 shell 的路径没有指向可执行文件（A 行）。

import {spawn} from 'node:child_process';

const childProcess = spawn(
 'echo hello',
 {
 stdio: ['inherit', 'inherit', 'pipe'],
 shell: '/bin/does-not-exist', // (A)
 }
);
childProcess.on('error', (err) => { // (B)
 assert.equal(
 err.toString(),
 'Error: spawn /bin/does-not-exist ENOENT'
 );
});

这是我们第一次使用事件来处理子进程。在 B 行，我们为'error'事件注册了一个事件监听器。当前代码片段完成后，子进程开始。这有助于防止竞争条件：当我们开始监听时，我们可以确保事件尚未被触发。

12.2.6.2 Shell 中发生了错误

如果 shell 代码包含错误，我们不会收到'error'事件（B 行），而是会收到一个带有非零退出代码的'exit'事件（A 行）：

import {Readable} from 'node:stream';
import {spawn} from 'node:child_process';

const childProcess = spawn(
 'does-not-exist',
 {
 stdio: ['inherit', 'inherit', 'pipe'],
 shell: true,
 }
);
childProcess.on('exit',
 async (exitCode, signalCode) => { // (A)
 assert.equal(exitCode, 127);
 assert.equal(signalCode, null);
 const stderr = Readable.toWeb(
 childProcess.stderr.setEncoding('utf-8'));
 assert.equal(
 await readableStreamToString(stderr),
 '/bin/sh: does-not-exist: command not found\n'
 );
 }
);
childProcess.on('error', (err) => { // (B)
 console.error('We never get here!');
});

12.2.6.3 进程被终止

如果在 Unix 上终止进程，退出代码是null（C 行），信号代码是一个字符串（D 行）：

import {Readable} from 'node:stream';
import {spawn} from 'node:child_process';

const childProcess = spawn(
 'kill $$', // (A)
 {
 stdio: ['inherit', 'inherit', 'pipe'],
 shell: true,
 }
);
console.log(childProcess.pid); // (B)
childProcess.on('exit', async (exitCode, signalCode) => {
 assert.equal(exitCode, null); // (C)
 assert.equal(signalCode, 'SIGTERM'); // (D)
 const stderr = Readable.toWeb(
 childProcess.stderr.setEncoding('utf-8'));
 assert.equal(
 await readableStreamToString(stderr),
 '' // (E)
 );
});

请注意，没有错误输出（E 行）。

子进程不是自己终止（A 行），我们也可以暂停它更长时间，然后通过我们在 B 行记录的进程 ID 手动终止它。

如果我们在 Windows 上杀死一个子进程会发生什么？

• exitCode 是 1。

• signalCode 是 null。

12.2.7 等待子进程退出

有时我们只想等到命令执行完毕。这可以通过事件和 Promise 来实现。

12.2.7.1 通过事件等待

import * as fs from 'node:fs';
import {spawn} from 'node:child_process';

const childProcess = spawn(
 `(echo first && echo second) > tmp-file.txt`,
 {
 shell: true,
 stdio: 'inherit',
 }
);
childProcess.on('exit', (exitCode, signalCode) => { // (A)
 assert.equal(exitCode, 0);
 assert.equal(signalCode, null);
 assert.equal(
 fs.readFileSync('tmp-file.txt', {encoding: 'utf-8'}),
 'first\nsecond\n'
 );
});

我们使用标准的 Node.js 事件模式，并为 'exit' 事件注册了一个监听器（A 行）。

12.2.7.2 通过 Promises 等待

import * as fs from 'node:fs';
import {spawn} from 'node:child_process';

const childProcess = spawn(
 `(echo first && echo second) > tmp-file.txt`,
 {
 shell: true,
 stdio: 'inherit',
 }
);

const {exitCode, signalCode} = await onExit(childProcess); // (A)

assert.equal(exitCode, 0);
assert.equal(signalCode, null);
assert.equal(
 fs.readFileSync('tmp-file.txt', {encoding: 'utf-8'}),
 'first\nsecond\n'
);

我们在 A 行使用的辅助函数 onExit() 返回一个 Promise，如果触发了 'exit' 事件，它就会被满足：

export function onExit(eventEmitter) {
 return new Promise((resolve, reject) => {
 eventEmitter.once('exit', (exitCode, signalCode) => {
 if (exitCode === 0) { // (B)
 resolve({exitCode, signalCode});
 } else {
 reject(new Error(
 `Non-zero exit: code ${exitCode}, signal ${signalCode}`));
 }
 });
 eventEmitter.once('error', (err) => { // (C)
 reject(err);
 });
 });
}

如果 eventEmitter 失败，返回的 Promise 被拒绝，await 在 A 行抛出异常。onExit() 处理两种失败情况：

• exitCode 不是零（B 行）。发生了这种情况：

  • 如果有 shell 错误。那么 exitCode 大于零。

  • 如果在 Unix 上杀死子进程。那么 exitCode 是 null，signalCode 是非空的。

    • 在 Windows 上杀死子进程会产生一个 shell 错误。

• 一个 'error' 事件被触发（C 行）。如果孩子进程无法被生成，就会发生这种情况。

12.2.8 终止子进程

12.2.8.1 通过 AbortController 终止子进程

在这个例子中，我们使用 AbortController 来终止一个 shell 命令：

import {spawn} from 'node:child_process';

const abortController = new AbortController(); // (A)

const childProcess = spawn(
 `echo Hello`,
 {
 stdio: 'inherit',
 shell: true,
 signal: abortController.signal, // (B)
 }
);
childProcess.on('error', (err) => {
 assert.equal(
 err.toString(),
 'AbortError: The operation was aborted'
 );
});
abortController.abort(); // (C)

我们创建一个 AbortController（A 行），将其信号传递给 spawn()（B 行），并通过 AbortController 终止 shell 命令（C 行）。

子进程是异步启动的（在当前代码片段执行后）。这就是为什么我们可以在进程甚至开始之前中止，以及为什么在这种情况下我们看不到任何输出。

12.2.8.2 通过 .kill() 终止子进程

在下一个例子中，我们通过方法 .kill() 终止一个子进程（最后一行）：

import {spawn} from 'node:child_process';

const childProcess = spawn(
 `echo Hello`,
 {
 stdio: 'inherit',
 shell: true,
 }
);
childProcess.on('exit', (exitCode, signalCode) => {
 assert.equal(exitCode, null);
 assert.equal(signalCode, 'SIGTERM');
});
childProcess.kill(); // default argument value: 'SIGTERM'

再次，在孩子进程开始之前我们就杀死了它（异步！），并且没有输出。

12.3 同步生成进程：spawnSync()

spawnSync(
 command: string,
 args?: Array<string>,
 options?: Object
): Object

spawnSync() 是 spawn() 的同步版本 - 它会等待子进程退出，然后同步返回一个对象。

参数大多与spawn()相同。options 有一些额外的属性 - 例如：

• .input: string | TypedArray | DataView

  如果这个属性存在，它的值将被发送到子进程的标准输入。

• .encoding: string（默认：'buffer'）

  指定用于所有标准 I/O 流的编码。

该函数返回一个对象。它最有趣的属性是：

• .stdout: Buffer | string

  包含写入子进程标准输出流的内容。

• .stderr: Buffer | string

  包含写入子进程标准错误流的内容。

• .status: number | null

  包含子进程的退出代码或 null。退出代码或信号代码中的一个是非空的。

• .signal: string | null

  包含孩子进程的信号代码或 null。退出代码或信号代码中的一个是非空的。

• .error?: Error

  只有在生成失败时才会创建这个属性，然后包含一个错误对象。

使用异步的 spawn() 时，子进程并行运行，我们可以通过流读取标准 I/O。相反，同步的 spawnSync() 收集流的内容并将其同步返回给我们（见下一小节）。

12.3.1 shell 命令何时执行？

使用同步的 spawnSync() 时，命令的子进程是同步启动的。以下代码演示了这一点：

import {spawnSync} from 'node:child_process';

spawnSync(
 'echo', ['Command starts'],
 {
 stdio: 'inherit',
 shell: true,
 }
);
console.log('After spawnSync()');

这是输出：

Command starts
After spawnSync()

12.3.2 从标准输出读取

以下代码演示了如何读取标准输出：

import {spawnSync} from 'node:child_process';

const result = spawnSync(
 `echo rock && echo paper && echo scissors`,
 {
 stdio: ['ignore', 'pipe', 'inherit'], // (A)
 encoding: 'utf-8', // (B)
 shell: true,
 }
);
console.log(result);
assert.equal(
 result.stdout, // (C)
 'rock\npaper\nscissors\n'
);
assert.equal(result.stderr, null); // (D)

在 A 行，我们使用 options.stdio 告诉 spawnSync() 我们只对标准输出感兴趣。我们忽略标准输入，并将标准错误传输到父进程。

因此，我们只能得到标准输出的结果属性（C 行），标准错误的属性是 null（D 行）。

由于我们无法访问spawnSync()内部使用的流来处理子进程的标准 I/O，我们通过options.encoding（B 行）告诉它使用哪种编码。

12.3.3 向子进程的 stdin 发送数据

我们可以通过选项属性.input（A 行）向子进程的标准输入流发送数据：

import {spawnSync} from 'node:child_process';

const result = spawnSync(
 `sort`,
 {
 stdio: ['pipe', 'pipe', 'inherit'],
 encoding: 'utf-8',
 input: 'Cherry\nApple\nBanana\n', // (A)
 }
);
assert.equal(
 result.stdout,
 'Apple\nBanana\nCherry\n'
);

12.3.4 处理不成功的退出（包括错误）

有三种主要的不成功的退出情况（当退出代码不为零时）：

• 子进程无法被生成。

• shell 中发生错误。

• 进程被终止。

12.3.4.1 子进程无法生成

如果生成失败，spawn()会发出一个'error'事件。相比之下，spawnSync()将result.error设置为一个错误对象：

import {spawnSync} from 'node:child_process';

const result = spawnSync(
 'echo hello',
 {
 stdio: ['ignore', 'inherit', 'pipe'],
 encoding: 'utf-8',
 shell: '/bin/does-not-exist',
 }
);
assert.equal(
 result.error.toString(),
 'Error: spawnSync /bin/does-not-exist ENOENT'
);

12.3.4.2 shell 中发生错误

如果在 shell 中发生错误，退出代码result.status大于零，result.signal为null：

import {spawnSync} from 'node:child_process';

const result = spawnSync(
 'does-not-exist',
 {
 stdio: ['ignore', 'inherit', 'pipe'],
 encoding: 'utf-8',
 shell: true,
 }
);
assert.equal(result.status, 127);
assert.equal(result.signal, null);
assert.equal(
 result.stderr, '/bin/sh: does-not-exist: command not found\n'
);