Stream API

可读流 API

ReadableStream

ReadableStream 可读流实例创建: new ReadableStream(underlyingSource, queuingStrategy)

• underlyingSource 参数（可选）: 该参数包含构造流对应的行为方法和属性的对象，主要包含如下

  • start(controller):
    • → 该行为方法会在可读流实例对象构造时立即调用，我们可以通过该方法来对流做一些操作（控制）
    • → 其中传递给该方法的 controller 就是对应的 ReadableStreamDefaultController | ReadableByteStreamController 实例对象（具体是那一个 controller 实例对象，取决于 type 属性的定义的值）
  • **pull(controller): **
    • → 该方法与上面的 start 方法类型，也是又开发者进行定义
    • → 不过该方法在流内部队列未满时，会重复调用该方法，知道队列补满
    • → 如果该方法返回的是一个 promise 时，那么会停止对应的重复调用该方法，会等待 promise 完成后再进行重复调用（但如果 promise 是 rejected 状态，该流就会出错）
    • → 其中传递给该方法的 controller 就是对应的 ReadableStreamDefaultController | ReadableByteStreamController 实例对象（具体是那一个 controller 实例对象，取决于 type 属性的定义的值）
    • 了解: 在 ReadableStream 的上下文中，pull 方法的使用通常与流的数据源需要按需或异步提供数据到流中有关 → 当流的消费者（如 ReadableStreamReader）需要更多数据时，并且数据的提供不是立即的（例如，数据来源于网络请求、文件读取、数据库查询等异步操作），此时就会用到pull方法
  • cancel(reason):
    • → 当该流被取消时会才会执行该方法（如调用了 ReadableStream.canel() 方法），开发者可以在该方法中对进行对流的访问进行释放或清除一些副作用等操作
    • → 如果开发者定义这个方法的逻辑时，是异步的，可以返回一个 promise 来表明对应的成功或失败
  • type: → 该属性用于控制正在处理可读类型的值，如果该值为 "bytes" 则传递的控制器对象 controller 就为 ReadableStreamByteStreamController（能够处理 body（带你自己的缓冲区）/字节流），反之就为 ReadableStreamDefaultController
  • autoAllocateChunkSize: MDN → 于字节流，开发人员可以使用正整数值设置 autoAllocateChunkSize 以打开流的自动分配功能 → 启用此功能后，流实现将自动分配一个具有给定整数大小的 ArrayBuffer，并调用底层源代码，就好像消费者正在使用 BYOB reader 一样

• queuingStrategy 参数（可选）: 该参数用于定义流中队列策略的对象，可以又如下两个参数 → queuingStrategy: 队列策略

  • highWaterMark: → 需要传入一个非负整数 → 该属性用于定义应用在背压之前可以包含在内部队列中块的总数数
  • size(chunk): → 一个包含 chunk 的方法 → 该方法用于表示每一个分块的使用的大小（以字节为单位）

• 返回值: ReadableStream 实例对象

• + 基本创建
  const stream = new ReadableStream({ // -- 1. 通过 new ReadableStream 创建对应的 ReadableStream 可读流实例对象// -- 2. 创建实例的第一个配置对象中的，start 方法，接收一个 controller 对象（默认为 ReadableStreamDefaultController 实例对象）start(controller) {/** 3. controller 对象的属性与方法+ desiredSize: 该属性返回填充满当前流队列需要的大小+ close: 该方法用于关闭一个流（不会丢弃对应的数据块，只是关闭对流的读取）+ enqueue: 该方法可以传入一个 chunk，用于将该 chunk 数据块压入到流的队列中+ error: 该方法可以传入一个 reason，用于控制流的出错（抛出错误）*/},pull(controller) { },cancel(reason) {// -- clear effect ....}
  })console.log(stream)// -- 可以通过 ReadableStreamDefaultReader 构成函数创建对应的流的 reader 并锁定该流（传入对应的 stream 可读流）
  const reader = new ReadableStreamDefaultReader(stream)
  console.log(reader)// -- 也可以通过 stream 可读流中的 getReader 方法获取对应的 reader 对象并锁定该流
  const reader = stream.getReader()
  console.log(reader)/** ↑ reader 实例对象的属性与方法+ closed: 该属性为一个 promise，当流关闭时兑现（调用 controller 对象中的 close 方法），当流抛出错误或 reader 锁被释放时拒绝兑现（调用 controller 中的 error 方法或调用 reader 中的 releaseLock 方法）+ cancel: 该方法用于取消对流的读取（完全结束: 所有数据会丢失，且流不再可读）+ read: 该方法返回一个 promies，且该 promise 兑现下一个分块（以供访问）+ releaseLock: 该方法用于释放 reader 对流的锁定*/
  

• + 简单示例
  const stream = new ReadableStream({ // -- 创建可读流实例对象start(controller) {// -- 通过 setTimeout 来模拟 "异步异步请求数据" 压入 chunk 到流的队列中controller.enqueue("k")setTimeout(() => {controller.enqueue("o")}, 1000)setTimeout(() => {controller.enqueue("n")}, 2000)setTimeout(() => {controller.enqueue("g")}, 3000)},cancel(reason) { // -- 当取消对流的读取时，就会触发该函数console.log(`Stream canceled: ${reason}`)}
  })// -- 读取 ↓
  const reader = new ReadableStreamDefaultReader(stream) // -- 创建对应 stream 可读流的 reader 读取器对象const read = () => { // -- 定义一个读取方法// -- 通过 reader 对象中的 read 方法来读取流中的分块内容 → 对应 promise 会兑现下一个分块的内容（迭代格式，具体开文档或笔记）reader.read().then(({ value, done }) => { // -- value 表示该分块的内容 : done 表示流是否已经将所有分块都读取了（以为分块数据可以继续读取）if (done) { // -- 当流已经读取完成时，打印对应提示并返回（中止该函数）console.log("Stream Completed")return}// -- 当还又分块可以读取时，打印对应分块内容并递归读取（已经读取的分块会被丢弃）console.log(value)textEl.textContent += valueread()}).catch(err => { // -- 错误处理console.log(err)})
  }const textEl = document.querySelector(".text") // -- 获取一个 html 元素用来展示流读取的结果read()// -- 🔺因为 read 方法返回的是 promise 对象，所以也可以通过 await 来使其同步处理对应的数据，下面的也是同理 → 此处暂时就不演示了...
  

• 从上面简单的示例中可以看出，如果服务器返回的数据是流时，我们可以通过一个流来进行分块读取展示...

实例属性:

• cloked: 返回当前可读流是否已被某个 reader 进行锁定

实例方法:

• getReader: 该方法可以创建一个 reader 对象，并将流锁定 tip: 只有该 reader 将流释放后，其它 reader 才能使用

• tee: 该方法可以对当前的可读流进行拷贝，返回一个数组（包含对应的两个 ReadableStream 实例对象分支的数组）

  • fetch("./temp.txt").then(res => {const stream = res.bodyconsole.log(stream.tee()) // log: [ReadableStream, ReadableStream]
    })
    

• pipeThrough: 该方法允许流可以通过一个转换流 TransformStream 来处理流中的数据，然后会将转换后的结果输出到一个新的可读流中（返回该新的可读流） → pipeThrough(transformStream,options) → 如下示例

  • 1. 创建一个流用于使用对应 pipeTrough 方法，并向器压入一些默认的户据const stream = new ReadableStream({start(controller) { // -- 可读流在初始化时，给该流队列压入一些 chunk 数据块controller.enqueue("Kong")setTimeout(() => controller.enqueue("Xiao"), 1000)setTimeout(() => controller.enqueue("Huang"), 2000)}})
    

  • 2. 创建已个转换流，其中在转换流中的 transform 方法中将对应的流转换的数据字符转换成大写字符const tranfer = new TransformStream({transform(chunk, controller) { // -- 对流中的数据块进行一些处理controller.enqueue(chunk.toUpperCase())}})
    

  • 3.🔺调用 stream 中的 pipeThrough 对该流进行转换，返回对应转换的新流 → 并读取里面的数据进行查看// -- 通过 stream 中的 pipeThrough 方法，对该流进行一些转换，返回一个新的可读流const streamTransformed = stream.pipeThrough(tranfer)// -- 读取该转换后的可读流const reader = streamTransformed.getReader()const read = () => {reader.read().then(({ value, done }) => {if (done) returnconsole.log(value)read()})}read()
    

  • transformStream: 一个转换流

  • options: 参数主要有如下配置项

    • + preventClose：如果设置为 true，则源 ReadableStream 关闭时不会关闭管道的 ReadableStream+ preventAbort：如果设置为 true，则源 ReadableStream 被中断时不会中断管道的 ReadableStream+ preventCancel：如果设置为 true，则取消源 ReadableStream 时不会取消管道的 ReadableStream+ signal：一个 AbortSignal，可以用来中断管道操作
      

  • pipeTo: pipeTo(writableStream,oprions)

    • 该方法可以将对应的 ReadableStream 可读流直接连接一个 WritableStream 可写流上

    • 通过该方法可以直接将流中的数据直接通过 "管道" 到另一个流中，无需显示的读取和写入每个数据块 → 如下示例

    • 1. 用于示例的可读流与可写流的创建
      // -- 可读流const stream = new ReadableStream({start(controller) {controller.enqueue("Kong")setTimeout(() => controller.enqueue("Xiao"), 1000)setTimeout(() => {controller.enqueue("Huang")controller.close() // -- 所有数据写入后，关闭可读流}, 2000)}})// -- 可写流const writableStream = new WritableStream({write(chunk, controller) {console.log(chunk)}})
      

    • 2.🔺通过 pipeTo 管道的方式直接将可读流中的数据写入可写流中（通过该方法，可以减少了显示读取与写入的操作）
      // 将上面的 stream 可读流的数据通过管道写入到 writableStreamstream.pipeTo(writableStream).then(() => {console.log('All data has been piped to the destination stream.')console.log(writableStream);}).catch((error) => {console.error('Piping failed:', error);})
      

    • writableStream: 一个目标的可读流对象

    • options: 主要有以下配置项

      • + preventClose：如果设置为 true，则源 ReadableStream 关闭时不会关闭目标 WritableStream+ preventAbort：如果设置为 true，则源 ReadableStream 被中断时不会中断目标 WritableStream+ signal：一个 AbortSignal，可以用来中断管道操作
        

ReadableStreamDefaultController

该 API 是一个控制器，通过该实例对下给你可以控制 ReaderStream 可读流的状态和内部队列 → 状态: 关闭可读流/抛出错误 | 内部队列: 压入流数据块入对应流的队列中

需要注意的是，该 API 没有构造函数（即可以理解为不能通过 new 来创建对应的实例对象），该实例会在 ReaderStream 实例构建时自动生成对应的 ReadableStreamDefaultController 实例对象

实例属性:

• desiredSize: 该属性返回流的内部队列填充满需要的大小 → 即前面内置队列概述中的计算公式的结果

实例方法:

• error: 该方法用于在 ReaderStream 上触发一个错误（如当流的源与到无法恢复的错误或内部自定义操作不通过或当某些某些无法预期的问题导致流无法继续提供数据或无法正常完成时，可以通过该方法来通知流的消费者） → error(reason)

• close: 该方法用于关闭对应关联的流 → close()

  • reader 将仍然可以从流中读取任何先前入队的数据块，如果你想完全的丢弃这个流并且丢弃任何入队的数据块

• enqueue: 该方法用于将给定的数据块压入到关联的流中 → enqueue(chunk)

该 API 具体的用法，在 ReadableStream 中进行示例

ReadableStreamDefaultReader

该 API 用于读取来自网络或外部资源提供的流数据（如 fetch 请求中的 response.body）的默认 render

实例创建: new ReadableStreamDefaultReader(stream) | 在一个 ReaderStream 上调用 getReader 方法（同样可以返回一个与其 stream 关联的 reader） → 对应可读流中的 locked 属性将为 true，表示该流当前的状态为锁定的

• const reader = new ReadableStreamDefaultReader(stream) // 创建方式1: 传入一个将要被读取的 ReadableStream 可读流const reader = stream.getReader() // -- 创建方式2: 也可以通过一个可读流上的 getRender 方法，来创建对应的 reader 并关联在该流上
  

实例属性:

• closed: 该属性返回一个 promise 对象，该 promise 会在流关闭时兑现（即 fulfilled 状态），当流抛出错误或 reader 锁被释放时拒绝兑现（即 rejected 状态） → 该属性可以使我们可以编写一个流在结束时的响应代码

  • reader.closed.then(() => {// -- 读取结束时进入该回调
    }).catch(() => {// -- 读取出现错误或释放 render 上的锁时进入该回调
    })
    

实例方法:

• read: 该方法返回一个 promise，并提供流的内部队列中的下一个分块进行访问（可通过递归读取流中所有的分块数据）

  • 返回值 promise 中提供的下一个分块内容格式: { value: theChunk, done: boolean } → 我们可以简单理解为该就为一个迭代器的返回值格式+ 如果分块可用时: { value: theChunk, done: false }+ 如果分块不可用时: { value: undefined, done: true }+ 如果流发生错误时: 该 promise 也将因相关错误拒绝兑现（即 rejected 状态）+ 其中: 上面的 value 是以一个 Uint8Array 类型来兑现 
    

  • 具体使用在 ReadableStream 在 ReadableStream 中进行示例

• cancel: 该方法返回一个 promise，该 promise 会在流取消时兑现（消费者在流中调用该方法发出取消流的信号）

  • cancel 用于在不再需要来自一个流的任何数据的情况下完全结束这个流，即使仍有排队等待的数据块
  • 调用 cancel 后该数据丢失，并且流不再可读
  • 如果想要仍然可以读这些数据块而不完全结束这个流，我们可以使用 ReaderStreamDefaultController,close() 方法

• release: 该方法用于释放 render 对流的锁定

该 API 具体的用法，在 ReadableStream 中进行示例