深入理解 JavaScript 中的 Streams API (ReadableStream, WritableStream, TransformStream) 及其应用场景。

大家好，我是你们今天的“数据流大法好”讲师，让我们一起潜入 JavaScript Streams API 的世界，看看这些“水管工”是如何优雅地处理数据的。

开场白：告别“一锤子买卖”的数据处理

在传统的 JavaScript 开发中，我们经常遇到这样的场景：一次性加载整个文件，然后一股脑地处理它。如果文件很小，那还好说，但如果是个 GB 级别的“巨无霸”，那就只能“呵呵”了。内存直接爆炸，浏览器直接卡死，用户体验直接跌到谷底。

想象一下，你要处理一个巨大的日志文件，里面记录了服务器的各种行为。传统的做法是，把整个文件读到内存里，然后开始疯狂地 split、substring、replace。这种做法就像一口气吃下一个巨大的汉堡，不仅撑得慌，而且消化不良。

JavaScript Streams API 就是来拯救我们的。它允许我们以更“流式”的方式处理数据，就像用水管传输水一样，一点一点地处理，而不是一次性把所有水都倒进来。

第一部分：Streams API 的核心概念

Streams API 是一套用于异步处理流式数据的接口。它定义了三种主要类型的流：

• ReadableStream (可读流): 数据源，负责产生数据。想象一下水龙头，它源源不断地流出水。
• WritableStream (可写流): 数据目的地，负责接收数据。想象一下水桶，它用来接收水。
• TransformStream (转换流): 数据转换器，负责处理数据，将一种格式的数据转换成另一种格式的数据。想象一下净水器，它把脏水变成干净的水。

这三者组合起来，就像一个数据处理的“流水线”，数据从 ReadableStream 流出，经过 TransformStream 的处理，最终流入 WritableStream。

1. ReadableStream：数据的“源头活水”

ReadableStream 代表一个可以从中读取数据的流。你可以从网络请求、文件、甚至是用户输入中创建 ReadableStream。

• 创建 ReadableStream：

const readableStream = new ReadableStream({
  start(controller) {
    controller.enqueue('Hello, ');
    controller.enqueue('World!');
    controller.close(); // 完成数据发送
  },
  pull(controller) {
      // 可选：当消费者需要更多数据时调用
      console.log("请求更多数据")
  },
  cancel(reason) {
      // 可选：当流被取消时调用
      console.log("流被取消", reason);
  }
});

这个例子创建了一个简单的 ReadableStream，它会依次发送 "Hello, " 和 "World!"。start 方法是流的初始化方法，controller.enqueue() 用于将数据添加到流中，controller.close() 用于表示流已经完成。 pull 方法是当消费者需要更多数据时调用，类似于“再来点”。cancel方法是在流被取消时调用。

• 读取 ReadableStream：

使用 getReader() 方法可以获得一个 Reader 对象，用于从流中读取数据。

const reader = readableStream.getReader();

async function readData() {
  while (true) {
    const { done, value } = await reader.read();
    if (done) {
      console.log('读取完成');
      break;
    }
    console.log('读取到的数据:', value);
  }
}

readData(); // 输出 "Hello, " 和 "World!"

reader.read() 方法会返回一个 Promise，resolve 的结果是一个对象，包含 done (表示是否读取完毕) 和 value (读取到的数据)。

• 常用方法和属性:

方法/属性	描述
getReader()	创建并返回一个 Reader 对象，用于从流中读取数据。
pipeTo(writable)	将 ReadableStream 的数据直接管道到 WritableStream，无需手动读取和写入。
pipeThrough(transform)	将 ReadableStream 的数据通过 TransformStream 转换后，管道到另一个 WritableStream (实际上是返回一个新的 ReadableStream，可以继续 .pipeTo())。
cancel(reason)	取消流的读取，可以传递一个原因 (reason)。
locked	只读属性，表示流是否被锁定 (是否有 Reader 正在使用)。

2. WritableStream：数据的“最终归宿”

WritableStream 代表一个可以写入数据的流。你可以将数据写入文件、网络连接，甚至是控制台。

• 创建 WritableStream：

const writableStream = new WritableStream({
  write(chunk) {
    // 处理写入的数据
    console.log('写入的数据:', chunk);
    return Promise.resolve(); // 表示写入成功
  },
  close() {
    console.log('写入完成');
  },
  abort(reason) {
    console.error('写入中止:', reason);
  }
});

write 方法用于处理写入的数据，close 方法在流关闭时调用，abort 方法在流中止时调用。write方法必须返回一个 Promise，表示写入操作是否成功。

• 写入 WritableStream：

使用 getWriter() 方法可以获得一个 Writer 对象，用于将数据写入流。

const writer = writableStream.getWriter();

async function writeData() {
  await writer.write('Hello, ');
  await writer.write('World!');
  await writer.close(); // 关闭流
}

writeData(); // 输出 "Hello, " 和 "World!"

writer.write() 方法用于写入数据，writer.close() 方法用于关闭流。

• 常用方法和属性:

方法/属性	描述
getWriter()	创建并返回一个 Writer 对象，用于将数据写入流。
abort(reason)	中止流的写入，可以传递一个原因 (reason)。
locked	只读属性，表示流是否被锁定 (是否有 Writer 正在使用)。

3. TransformStream：数据的“变形金刚”

TransformStream 可以在读取和写入之间转换数据。它本质上是一个 ReadableStream 和一个 WritableStream 的组合，数据先写入 WritableStream，经过转换后，从 ReadableStream 中读取。

• 创建 TransformStream：

const transformStream = new TransformStream({
  transform(chunk, controller) {
    // 转换数据
    const transformedChunk = chunk.toUpperCase();
    controller.enqueue(transformedChunk);
  },
  flush(controller) {
    // 可选：在流关闭前最后一次处理数据
    console.log("Flush called!");
  }
});

transform 方法用于转换数据，controller.enqueue() 用于将转换后的数据添加到流中。flush方法是在流关闭前最后一次处理数据。

• 使用 TransformStream：

const readable = new ReadableStream({
  start(controller) {
    controller.enqueue('hello');
    controller.enqueue('world');
    controller.close();
  }
});

const writable = new WritableStream({
  write(chunk) {
    console.log('写入的数据:', chunk);
  }
});

readable.pipeThrough(transformStream).pipeTo(writable); // 输出 "HELLO" 和 "WORLD"

pipeThrough() 方法将 ReadableStream 的数据通过 TransformStream 转换后，管道到 WritableStream。

• 常用方法和属性:

TransformStream 本身没有太多独特的方法或属性，因为它本质上就是 ReadableStream 和 WritableStream 的组合。 你主要使用它的 readable 和 writable 属性来访问内部的 ReadableStream 和 WritableStream。

第二部分：Streams API 的应用场景

Streams API 在处理大型数据集、实时数据流、以及需要复杂数据转换的场景中非常有用。

1. 处理大型文件

假设你需要处理一个巨大的 CSV 文件，并统计其中某个字段的总和。使用 Streams API 可以避免一次性将整个文件加载到内存中。

async function processLargeCSV(file) {
  const decoder = new TextDecoder(); // 用于解码 ArrayBuffer
  let total = 0;

  const readableStream = file.stream(); // 从 File 对象创建 ReadableStream
  const transformStream = new TransformStream({
    transform(chunk, controller) {
      // chunk 是 Uint8Array
      const text = decoder.decode(chunk);
      const lines = text.split('n');

      for (const line of lines) {
        if (line.trim() === '') continue; // 忽略空行

        const fields = line.split(',');
        const value = parseFloat(fields[2]); // 假设第三个字段是需要统计的数值

        if (!isNaN(value)) {
          total += value;
        }
      }
    }
  });

  await readableStream.pipeThrough(transformStream).pipeTo(new WritableStream());

  console.log('总和:', total);
}

// 假设你有一个 <input type="file"> 元素
const fileInput = document.getElementById('fileInput');
fileInput.addEventListener('change', (event) => {
  const file = event.target.files[0];
  processLargeCSV(file);
});

这个例子首先从 File 对象创建一个 ReadableStream，然后创建一个 TransformStream 来解析 CSV 文件，提取需要统计的字段，并计算总和。最后，将结果输出到控制台。注意这里用到了 TextDecoder 来将 Uint8Array 转换为字符串。

2. 实时数据流处理

Streams API 非常适合处理实时数据流，例如来自 WebSocket 连接的数据。

const socket = new WebSocket('wss://example.com/data');

const readableStream = new ReadableStream({
  start(controller) {
    socket.addEventListener('message', (event) => {
      controller.enqueue(event.data); // 将收到的数据添加到流中
    });

    socket.addEventListener('close', () => {
      controller.close(); // 关闭流
    });

    socket.addEventListener('error', (error) => {
      controller.error(error); // 报告错误
    });
  },
  cancel() {
    socket.close(); // 关闭 WebSocket 连接
  }
});

const writableStream = new WritableStream({
  write(chunk) {
    // 处理接收到的数据
    console.log('接收到的数据:', chunk);
  }
});

readableStream.pipeTo(writableStream);

这个例子创建了一个 ReadableStream，它从 WebSocket 连接接收数据，并将其传递给 WritableStream 进行处理。

3. 数据压缩和解压缩

Streams API 可以与压缩和解压缩算法结合使用，以实现高效的数据传输。

// 假设你有一个压缩算法 (例如 gzip) 的实现
async function compressData(data) {
  const encoder = new TextEncoder();
  const compressedData = gzip(encoder.encode(data)); // 假设 gzip 函数存在
  return compressedData;
}

async function decompressData(compressedData) {
  const decompressedData = ungzip(compressedData); // 假设 ungzip 函数存在
  const decoder = new TextDecoder();
  const data = decoder.decode(decompressedData);
  return data;
}

const transformStream = new TransformStream({
  async transform(chunk, controller) {
    const compressedChunk = await compressData(chunk);
    controller.enqueue(compressedChunk);
  },
  async flush(controller) {
      //在流关闭前做一些清理工作
      console.log("执行flush");
  }
});

const readable = new ReadableStream({
  start(controller) {
    controller.enqueue('This is a long string of data that needs to be compressed.');
    controller.close();
  }
});

const writable = new WritableStream({
  write(chunk) {
    console.log('压缩后的数据:', chunk);
    decompressData(chunk).then(decompressed => console.log("解压后的数据：", decompressed))
  }
});

readable.pipeThrough(transformStream).pipeTo(writable);

这个例子创建了一个 TransformStream，它使用 gzip 算法压缩数据，并将压缩后的数据传递给 WritableStream。解压过程在 WritableStream 的 write 方法中进行。需要注意的是，这里只是一个示例，你需要自己实现 gzip 和 ungzip 函数，或者使用现有的压缩库。

第三部分：实战演练：一个简单的 HTTP 请求处理

让我们用 Streams API 来实现一个简单的 HTTP 请求处理。我们将使用 fetch API 获取数据，然后使用 ReadableStream 和 WritableStream 将数据写入控制台。

async function fetchAndLog(url) {
  const response = await fetch(url);

  if (!response.ok) {
    throw new Error(`HTTP error! Status: ${response.status}`);
  }

  const readableStream = response.body; // response.body 是一个 ReadableStream

  const writableStream = new WritableStream({
    write(chunk) {
      const decoder = new TextDecoder();
      const text = decoder.decode(chunk);
      console.log(text);
    }
  });

  await readableStream.pipeTo(writableStream);
}

fetchAndLog('https://httpbin.org/stream/10'); // 请求一个返回10行数据的API

这个例子使用 fetch API 获取数据，response.body 就是一个 ReadableStream。我们将这个 ReadableStream 管道到我们自定义的 WritableStream，它会将接收到的数据写入控制台。

第四部分：注意事项和最佳实践

• 错误处理： 在使用 Streams API 时，一定要注意错误处理。 ReadableStream、WritableStream 和 TransformStream 都有 cancel 和 abort 方法，用于处理错误和中止流。
• 背压 (Backpressure)： 背压是指当数据源 (ReadableStream) 的生产速度超过数据目的地 (WritableStream) 的消费速度时，需要采取一些措施来防止数据丢失。 Streams API 提供了背压机制，允许 WritableStream 向 ReadableStream 发出信号，告知其降低生产速度。 pull 方法就是用来处理背压的。
• 资源管理： 在使用 Streams API 时，要确保及时关闭流，释放资源。 尤其是处理文件和网络连接时，一定要在流完成或中止后关闭连接。
• 选择合适的 Stream 类型： 根据不同的场景选择合适的 Stream 类型。 如果只需要读取数据，就使用 ReadableStream；如果只需要写入数据，就使用 WritableStream；如果需要转换数据，就使用 TransformStream。
• 理解 Uint8Array： Streams API 经常处理 Uint8Array 类型的数据，这是 JavaScript 中表示二进制数据的常用方式。 你可以使用 TextEncoder 和 TextDecoder 在字符串和 Uint8Array 之间进行转换。

总结：拥抱“水流般”的数据处理方式

JavaScript Streams API 提供了一种强大而灵活的方式来处理流式数据。 掌握 Streams API，可以让你编写更高效、更可靠的代码，处理更复杂的数据处理任务。

记住，数据就像水，Streams API 就是你的水管，合理地利用它们，才能让数据顺畅地流动起来！