Node.js流技术问询：未连可写流时对象排队及流重启问题

问题1：Node.js的流是否原生支持在未连接到Writable流时对对象进行排队？ #

原生Readable流自带基于highWaterMark的缓存机制：当流未被消费（未pipe到Writable、未监听data事件或调用read()）时，push()传入的数据会被缓存，直到达到highWaterMark阈值后，push()会返回false，提示暂时停止写入。

但原生Readable没有提供像你实现的add()这类直接向队列添加数据的方法，它的设计逻辑是让开发者在_read()方法中按需生成数据。如果需要主动向流中添加数据并在未被消费时排队，你当前的实现思路是合理的，但要注意和原生缓存逻辑的配合。

问题2：调用super.push(null)后无法再处理数据，能否重启该流？ #

不能。调用push(null)会将Readable流标记为已结束（ended），流的状态一旦进入结束状态，就无法恢复或重启，后续再调用push()传入任何数据都会被忽略。

如果需要流在队列空时暂停输出，等有新数据进来再继续，不要调用push(null)，而是在_read()中当队列无数据时直接返回，等新数据通过add()加入队列后，手动触发推送逻辑即可。

对你实现的OrderedQueue类的优化建议 #

你当前的_read()方法存在问题：每次调用都会从队列取一个元素，空队列时直接push(null)，这会导致流直接结束，后续调用add()添加的数据无法被输出。

修改后的代码调整了_read()逻辑，并在add()中触发推送，同时遵循原生流的缓存机制：

import { Readable, ReadableOptions } from 'node:stream'

export class OrderedQueue<EventType = unknown> extends Readable {
  queue: EventType[] = []

  constructor(opts?: ReadableOptions) {
    super({ objectMode: true, highWaterMark: 1024, ...opts })
  }

  add(event: EventType): boolean {
    this.queue.push(event)
    // 当流处于可读取状态时，主动触发数据推送
    this._read()
    return this.queue.length <= this.readableHighWaterMark
  }

  _read(size: number): void {
    // 循环推送数据，直到队列空或缓存达到highWaterMark
    while (this.queue.length > 0) {
      const event = this.queue.shift()
      // push返回false说明缓存已满，停止推送
      if (!this.push(event)) {
        break
      }
    }
    // 队列空时不推送null，等待新数据加入
  }
}

内容的提问来源于stack exchange，提问作者Nick Grealy