JS `ReadableStream` 与 `WritableStream`：Web Streams API 的背压控制与管道操作

各位靓仔靓女，早上好！今天咱们聊聊 Web Streams API 里的两位重量级选手：ReadableStream 和 WritableStream。这哥俩可不是摆设，它们是浏览器里处理数据流的利器，特别是搞音视频、文件上传下载、网络通信的时候，有了它们，效率嗖嗖地！

咱们今天要深入探讨它们的背压控制和管道操作，保证让你听完之后，感觉自己也能轻松驾驭数据流。

啥是 Web Streams API？

简单来说，Web Streams API 是一套用于处理流式数据的 JavaScript API。它允许你异步地读取和写入数据块，而不用一次性把所有数据都加载到内存里。这就像你用水管往水桶里灌水，你可以控制水流的速度，而不是一次性把水库的水都倒进去。

主角登场：ReadableStream 和 WritableStream

ReadableStream (可读流): 顾名思义，用来读取数据的。你可以把它想象成一个水龙头，源源不断地流出数据。
WritableStream (可写流): 用来写入数据的。你可以把它想象成一个水桶，用来接收数据。

背压控制：数据流的交通规则

背压（backpressure）是 Web Streams API 里一个非常重要的概念。它指的是当一个流（比如 WritableStream）无法处理接收到的数据时，它可以通知上游的流（比如 ReadableStream）放慢速度，避免数据积压。

想象一下，你往水桶里灌水，如果水桶满了，你肯定要关小水龙头，不然水就要溢出来了。背压控制就是起到这个作用。

为啥需要背压控制？

避免内存溢出: 如果 WritableStream 处理数据的速度赶不上 ReadableStream 产生数据的速度，就会导致数据积压在内存里，最终可能导致内存溢出。
提高性能: 通过背压控制，可以让数据流的速度与处理能力相匹配，避免不必要的资源浪费。

ReadableStream 的背压机制

ReadableStream 会通过内部的队列来缓存数据。当队列满了，它会通知上游的 WritableStream 暂停发送数据。

getReader() 方法: ReadableStream 提供了 getReader() 方法来获取一个 ReadableStreamReader 对象，通过这个对象你可以读取数据。
read() 方法: ReadableStreamReader 对象的 read() 方法用于从流中读取数据。它返回一个 Promise，Promise resolve 的时候会返回一个对象，包含 value (读取到的数据) 和 done (是否读取完毕) 两个属性。
cancel() 方法: ReadableStreamReader 对象的 cancel() 方法用于取消读取流。
releaseLock() 方法: ReadableStreamReader 对象的 releaseLock()方法释放读取器的锁，允许其他读取器访问流。

WritableStream 的背压机制

WritableStream 通过 write() 方法和 close() 方法来控制数据的写入。

getWriter() 方法: WritableStream 提供了 getWriter() 方法来获取一个 WritableStreamDefaultWriter 对象，通过这个对象你可以写入数据。
write() 方法: WritableStreamDefaultWriter 对象的 write() 方法用于向流中写入数据。它返回一个 Promise，Promise resolve 的时候表示数据已经成功写入。如果 WritableStream 内部的队列满了，write() 方法返回的 Promise 会 pending，直到队列有空闲空间。
close() 方法: WritableStreamDefaultWriter 对象的 close() 方法用于关闭流。
abort() 方法: WritableStreamDefaultWriter 对象的 abort() 方法用于强制终止流。
releaseLock() 方法: WritableStreamDefaultWriter 对象的 releaseLock()方法释放写入器的锁，允许其他写入器访问流。

代码示例：模拟背压

async function simulateBackpressure() {
  let counter = 0;
  const readableStream = new ReadableStream({
    start(controller) {
      function pushData() {
        if (counter >= 10) {
          controller.close();
          return;
        }

        const data = `Data chunk ${counter}`;
        console.log(`ReadableStream: Pushing data - ${data}`);
        controller.enqueue(data);
        counter++;

        // 模拟生产数据的速度
        setTimeout(pushData, 500); // 每隔 500 毫秒推送一个数据块
      }

      pushData();
    },
    cancel(reason) {
      console.log(`ReadableStream: Cancellation reason - ${reason}`);
    },
  });

  const writableStream = new WritableStream({
    write(chunk) {
      return new Promise((resolve) => {
        console.log(`WritableStream: Processing chunk - ${chunk}`);

        // 模拟处理数据的速度，故意比生产数据慢
        setTimeout(() => {
          console.log(`WritableStream: Processed chunk - ${chunk}`);
          resolve();
        }, 1000); // 每隔 1000 毫秒处理一个数据块
      });
    },
    close() {
      console.log("WritableStream: All data chunks written and stream closed.");
    },
    abort(error) {
      console.error("WritableStream: Aborted", error);
    },
  });

  readableStream.pipeTo(writableStream).catch((error) => {
    console.error("Piping failed", error);
  });
}

simulateBackpressure();

在这个例子中，ReadableStream 每隔 500 毫秒生产一个数据块，而 WritableStream 每隔 1000 毫秒处理一个数据块。由于 WritableStream 处理数据的速度慢于 ReadableStream 生产数据的速度，就会产生背压。你可以通过控制台的输出来观察背压的效果。

管道操作：数据流的流水线

管道（piping）是 Web Streams API 里另一个重要的概念。它允许你把一个 ReadableStream 的输出直接连接到另一个 WritableStream 的输入，就像一条流水线一样。

pipeTo() 方法: ReadableStream 提供了 pipeTo() 方法来实现管道操作。

代码示例：管道操作

async function pipeExample() {
  const readableStream = new ReadableStream({
    start(controller) {
      for (let i = 0; i < 5; i++) {
        controller.enqueue(`Data chunk ${i}`);
      }
      controller.close();
    },
  });

  const writableStream = new WritableStream({
    write(chunk) {
      console.log(`WritableStream: Received chunk - ${chunk}`);
      return Promise.resolve();
    },
    close() {
      console.log("WritableStream: All data chunks written.");
    },
  });

  // 将 readableStream 的输出管道到 writableStream 的输入
  try {
    await readableStream.pipeTo(writableStream);
    console.log("Piping completed successfully.");
  } catch (error) {
    console.error("Piping failed:", error);
  }
}

pipeExample();

在这个例子中，readableStream 的输出直接管道到 writableStream 的输入，writableStream 会接收到 readableStream 产生的所有数据块。

更复杂的管道：中间件

你可以把多个流连接起来，构建更复杂的管道。这就像在流水线上添加多个工序，每个工序负责处理一部分数据。

async function complexPipeExample() {
  const readableStream = new ReadableStream({
    start(controller) {
      for (let i = 0; i < 5; i++) {
        controller.enqueue(`Raw data ${i}`);
      }
      controller.close();
    },
  });

  // 中间件 1：转换数据格式
  const transformStream1 = new TransformStream({
    transform(chunk, controller) {
      const transformedChunk = `Transformed: ${chunk}`;
      controller.enqueue(transformedChunk);
    },
  });

  // 中间件 2：添加时间戳
  const transformStream2 = new TransformStream({
    transform(chunk, controller) {
      const timestamp = new Date().toISOString();
      const timestampedChunk = `${chunk} - ${timestamp}`;
      controller.enqueue(timestampedChunk);
    },
  });

  const writableStream = new WritableStream({
    write(chunk) {
      console.log(`WritableStream: Received chunk - ${chunk}`);
      return Promise.resolve();
    },
    close() {
      console.log("WritableStream: All data chunks written.");
    },
  });

  // 管道连接：readableStream -> transformStream1 -> transformStream2 -> writableStream
  try {
    await readableStream
      .pipeThrough(transformStream1)
      .pipeThrough(transformStream2)
      .pipeTo(writableStream);
    console.log("Complex piping completed successfully.");
  } catch (error) {
    console.error("Complex piping failed:", error);
  }
}

complexPipeExample();

在这个例子中，我们使用了两个 TransformStream 作为中间件，分别负责转换数据格式和添加时间戳。readableStream 的输出先经过 transformStream1 处理，然后再经过 transformStream2 处理，最后才到达 writableStream。

TransformStream：数据流的变形金刚

TransformStream 是 Web Streams API 里一个非常灵活的类。它允许你对数据流进行各种各样的转换，比如格式转换、数据过滤、数据加密等等。

transform() 方法: TransformStream 的 transform() 方法用于对数据块进行转换。它接收两个参数：chunk (要转换的数据块) 和 controller (一个 TransformStreamDefaultController 对象)。
flush() 方法: TransformStream 的 flush() 方法用于在流关闭之前进行最后的处理。

表格总结

特性	`ReadableStream`	`WritableStream`	`TransformStream`
功能	读取数据	写入数据	转换数据
主要方法	`getReader()`, `pipeTo()`	`getWriter()`	`transform()`, `flush()`
背压控制	内部队列	`write()` 的 Promise	可以同时控制读写
使用场景	文件读取, 网络请求	文件写入, 网络响应	数据格式转换, 数据过滤

实际应用场景

音视频处理: 你可以使用 Web Streams API 来处理音视频流，实现实时播放、录制、转码等功能。
文件上传下载: 你可以使用 Web Streams API 来实现大文件的分块上传下载，提高效率和稳定性。
网络通信: 你可以使用 Web Streams API 来处理 WebSocket 数据流，实现实时通信功能。
数据压缩解压缩: 你可以使用 Web Streams API 和 Compression Streams API 来实现数据的压缩和解压缩。

注意事项

错误处理: 在处理数据流的时候，一定要注意错误处理。可以使用 try...catch 语句来捕获异常，并使用 abort() 方法来终止流。
资源释放: 在使用完流之后，一定要记得释放资源。可以使用 releaseLock() 方法来释放读取器或写入器的锁，并使用 close() 方法来关闭流。
浏览器兼容性: Web Streams API 的浏览器兼容性还不是很好，需要使用 polyfill 来兼容旧版本的浏览器。

总结

ReadableStream 和 WritableStream 是 Web Streams API 的核心组件，它们提供了强大的数据流处理能力。通过背压控制和管道操作，你可以构建高效、稳定的数据流应用。希望今天的讲解能让你对 Web Streams API 有更深入的了解，并在实际项目中灵活运用。

今天的讲座就到这里，感谢大家的聆听！如果有什么问题，欢迎随时提问。下次再见！

发表回复 取消回复

发表回复取消回复