node.js - _read() 未在可读流上实现

Question

本题是如何真正实现一个可读流的read方法。

我有一个可读流的实现:

import {Readable} from "stream";
this.readableStream = new Readable();

我收到这个错误

events.js:136 throw er; // Unhandled 'error' event ^

Error [ERR_STREAM_READ_NOT_IMPLEMENTED]: _read() is not implemented at Readable._read (_stream_readable.js:554:22) at Readable.read (_stream_readable.js:445:10) at resume_ (_stream_readable.js:825:12) at _combinedTickCallback (internal/process/next_tick.js:138:11) at process._tickCallback (internal/process/next_tick.js:180:9) at Function.Module.runMain (module.js:684:11) at startup (bootstrap_node.js:191:16) at bootstrap_node.js:613:3

错误发生的原因很明显，我们需要这样做:

  this.readableStream = new Readable({
      read(size) {
        return true;
      }
    });

虽然我不太明白如何实现读取方法。

唯一有用的就是调用

this.readableStream.push('some string or buffer');

如果我尝试做这样的事情:

   this.readableStream = new Readable({
          read(size) {
            this.push('foo');   // call push here!
            return true;
          }
     });

然后什么也没有发生 - 可读性没有任何结果!

此外，这些文章说您不需要实现 read 方法:

https://github.com/substack/stream-handbook#creating-a-readable-stream

https://medium.freecodecamp.org/node-js-streams-everything-you-need-to-know-c9141306be93

我的问题是 - 为什么在 read 方法中调用 push 什么都不做？唯一对我有用的就是在别处调用 readable.push()。

Answer 1

why does calling push inside the read method do nothing? The only thing that works for me is just calling readable.push() elsewhere.

我认为这是因为您没有使用它，您需要将它通过管道传输到可写流(例如 stdout)或者只是通过 data 事件使用它:

const { Readable } = require("stream");

let count = 0;
const readableStream = new Readable({
    read(size) {
        this.push('foo');
        if (count === 5) this.push(null);
        count++;
    }
});

// piping
readableStream.pipe(process.stdout)

// through the data event
readableStream.on('data', (chunk) => {
  console.log(chunk.toString());
});

它们都应该打印 5 次 foo(尽管它们略有不同)。您应该使用哪一个取决于您要实现的目标。

Furthermore, these articles says you don't need to implement the read method:

你可能不需要它，这应该有效:

const { Readable } = require("stream");

const readableStream = new Readable();

for (let i = 0; i <= 5; i++) {
    readableStream.push('foo');
}
readableStream.push(null);

readableStream.pipe(process.stdout)

在这种情况下，您无法通过 data 事件捕获它。此外，这种方式不是很有用且效率不高，我想说的是，我们只是一次将所有数据推送到流中(如果它很大，所有内容都会在内存中)，然后使用它。