Stream API - Readable Streams
在 Stream API 中，readable stream 是一种数据源，我们可以从中以顺序和异步的方式读取数据。它是获取底层源数据的一种标准化方式。底层源是网络上存在的资源，它们分为以下两种类型 −
Push source − 在这种类型中，当你访问它们时数据会被推送到你这里。你可以控制 stream，比如何时开始、何时暂停甚至何时终止当前 stream。例如，视频游戏流媒体。
Pull source − 在这种类型中，你需要明确地从它们请求数据。例如，通过 Fetch 或 XHR 调用访问文件。
在 Readable Stream 中，数据以小块（chunk）的形式存在，因此它被顺序读取，每次一块。一个 chunk 可以是一个字节，也可以是更大的大小。因此，stream 中的 chunk 大小可能不同。现在让我们了解 readable stream 的工作原理。
Readable Stream 的工作原理
Readable stream 的工作原理非常简单明了。在 readable stream 中，数据块被放入队列（enqueue）中。这意味着这些块正在队列中等待读取。这里还有另一个内部队列，用于跟踪未读的块。块由 reader 读取。它一次处理一个块的数据，并允许你对数据执行操作。一个 reader 一次只能读取一个 stream。当 reader 开始读取 stream 时，该 stream 会被锁定给该 reader，意味着其他 reader 不能读取该 stream。如果你想让另一个 reader 读取该 stream，则需要终止第一个 reader 或创建 tee stream。此外，每个 reader 都有自己的 controller，可以控制 stream 的开始、关闭或暂停。
它还有一个 consumer，负责处理从 readable stream 接收到的数据，进行处理并执行操作。
Readable Stream 接口
Stream API 支持三种 readable stream 接口 −
ReadableStream 接口
ReadableStreamDefaultReader 接口
ReadableStreamDefaultController 接口
ReadableStream 接口
ReadableStream 接口用于表示数据流的可读 stream。它通常与 Fetch API 一起使用来处理响应流。它也可以处理开发者定义的 stream 的响应流。
构造函数
为了为给定的 handlers 创建 readable stream 对象，ReadableStream 接口提供了 ReadableStream() 构造函数。
语法
const newRead = new ReadableStream() Or const newRead = new ReadableStream(UnderlyingSource) Or const newRead = new ReadableStream(UnderlyingSource, QueuingStrategy)
以下是 ReadableStream() 构造函数的可选参数 −
UnderlyingSource − 此对象提供了定义 stream 实例行为的各种方法和属性。方法包括：start()、pull() 和 cancel()，属性包括：type 和 autoAllocateChunkSize。
QueuingStrategy − 此对象用于定义给定 stream 的排队策略。它接受两个参数：highWaterMark 和 size(chunk)。
实例属性
ReadableStream 接口提供的属性是只读属性。因此，ReadableStream 提供的属性为 −
| Sr.No. | Property & Description |
|---|---|
| 1 | ReadableStream.locked 此属性用于检查 readable stream 是否被锁定给某个 reader。 |
方法
以下是 ReadableStream 接口常用的方法 −
| Sr.No. | Method & Description |
|---|---|
| 1 | ReadableStream.cancel() 该方法返回一个 promise，当流被取消时该 promise 将 resolve。 |
| 2 | ReadableStream.getReader() 该方法用于创建 reader 并将流锁定到该 reader 上。在该 reader 释放之前，不允许其他 reader 使用。 |
| 3 | ReadableStream.pipeThrough() 该方法用于以可链式的方式将当前流通过 transform stream 进行管道传输。 |
| 4 | ReadableStream.pipeTo() 该方法用于将当前 ReadableStream 管道传输到指定的 WriteableStream。当管道传输过程成功完成时返回 promise，或因某些错误而 reject。 |
| 5 | ReadableStream.tee() 该方法用于获取一个包含两个新 ReadableStream 对象作为结果分支的两元素数组。 |
ReadableStreamDefaultReader 接口
ReadableStreamDefaultReader 接口用于表示默认 reader，它将从网络读取流数据。也可以从 ReadableStream 读取。
构造函数
ReadableStreamDefaultReader 接口提供 ReadableStreamDefaultReader() 构造函数来创建 readableStreamDefaultReader 对象。
语法
const newRead = new ReadableStreamDefaultReader(myStream)
该构造函数仅包含一个参数，即 myStream。它将读取 ReadableStream。
实例属性
ReadableStreamDefaultReader 接口提供的属性是只读属性。因此 ReadableStreamDefaultReader 提供的属性为 −
| Sr.No. | Property & Description |
|---|---|
| 1 | ReadableStreamDefaultReader.closed 该属性返回一个 promise，当流关闭时 resolve，或因某些错误而 reject。它允许您编写在流传输过程结束时响应的程序。 |
方法
以下是 ReadableStream 接口常用的方法 −
| Sr.No. | Method & Description |
|---|---|
| 1 | ReadableStreamDefaultReader.cancel() 该方法返回一个 promise，当流被取消时该 promise 将 resolve。 |
| 2 | ReadableStreamDefaultReader.read() 该方法返回一个 promise，提供对流队列中下一个 chunk 或片段的访问。 |
| 3 | ReadableStreamDefaultReader.releaseLock() 该方法用于移除 reader 对流的锁定。 |
ReadableStreamDefaultController 接口
ReadableStreamDefaultController 接口表示一个控制器，允许我们控制 ReadableStream 的状态或内部队列。它不提供任何控制器，实例在构造 ReadableStream 时自动创建。
实例属性
| Sr.No. | Property & Description |
|---|---|
| 1 | ReadableStreamDefaultController.desiredSize 该属性用于查找填充流内部队列所需的期望大小。 |
ReadableStreamDefaultController 接口提供的属性是只读属性。因此 ReadableStreamDefaultController 提供的属性为 −
方法
以下是 ReadableStreamDefaultController 接口常用的方法 −
| Sr.No. | Property & Description |
|---|---|
| 1 | ReadableStreamDefaultController.close() 该方法用于关闭相关的流。 |
| 2 | ReadableStreamDefaultController.enqueue() 该方法用于在相关的流中入队指定的 chunk 或片段。 |
| 3 | ReadableStreamDefaultController.error() 该方法将导致与相关流的任何未来交互发生错误。 |
示例 - 创建 ReadableStream
在下面的程序中，我们将使用 ReadableStream 构造函数创建一个自定义的可读流。首先，我们创建一个函数，用于分块生成数据。然后，我们使用 ReadableStream() 构造函数创建一个可读流，其中包含 start() 函数。该 start() 函数使用 pData() 递归函数，通过 controller 将 myData() 函数的数据推送到消费者，每次推送操作之间设置 1 秒的超时。现在，我们使用 getReader() 函数创建一个 reader 来消费流中的数据。然后，我们创建一个 readMyData() 函数，通过 reader 递归读取流中的数据。当流结束时，done 标志被设置为 true，我们退出递归循环。
<!DOCTYPE html>
<html>
<body>
<script>
// 为流生成数据的函数
function* myData() {
yield 'pink';
yield 'blue';
yield 'yellow';
yield 'green';
}
// 使用 ReadableStream() 函数创建可读流
const readStream = new ReadableStream({
start(controller) {
const data = myData();
// 将数据添加到流中
function pData() {
const { done, value } = data.next();
if (done) {
// 如果没有更多数据可用，则关闭流
controller.close();
return;
}
// 将数据推送到消费者
controller.enqueue(value);
// 1 秒后继续推送数据
setTimeout(pData, 1000);
}
// 调用 pData 函数开始推送数据
pData();
}
});
// 为可读流创建 reader
const myreader = readStream.getReader();
function readMyData() {
myreader.read().then(({ done, value }) => {
if (done) {
// 流已关闭
console.log('Stream is closed');
return;
}
// 处理接收到的数据
console.log('Received data:', value);
// 继续读取数据
readMyData();
});
}
// 调用 readMyData() 函数开始
// 从可读流中读取数据
readMyData();
</script>
</body>
</html>
输出
结论
这就是 Stream API 中的可读流。它们是 Stream API 中最重要且最常用的流类型。它们几乎被所有网络浏览器支持，如 Chrome、Firefox、Opera、Edge、Safari 等。在下一篇文章中，我们将学习 Stream API 的可写流。