Node.js中readable._read()调用位置溯源及按需调用机制解析

Great question—this scenario cuts right to the core of how Node.js Readable streams handle demand-driven data fetching and backpressure. Let’s break down the behavior first, then dive into the relevant parts of the Node.js source code to see exactly where those _read() calls are triggered.

First, Let’s Recap Your Stream’s Behavior #

Your custom Readable stream pushes one character (1 byte) at a time with a 100ms delay. When you pipe it to head -c5, here’s the step-by-step play:

• head only needs 5 bytes of data, so it signals your stream to stop once that’s delivered.
• Each time your _read() method pushes a byte, the stream checks if the total data sent so far meets head’s demand. Since each push only covers 1 of the 5 needed bytes, the stream calls _read() again to fetch the next one.
• This repeats 5 times until head has its 5 bytes, then it closes the output stream, and your Readable stops calling _read().

Diving Into Node.js Source Code #

The core logic for Readable streams lives in lib/internal/streams/readable.js. Let’s walk through the key parts that trigger _read() in your scenario:

1. The read() Method: Initiating _read() #

When a consumer (like head via process.stdout) needs data, it indirectly calls the Readable stream’s internal read() method. This method checks if the stream’s internal buffer has enough data to satisfy the demand. If not:

• It sets state.reading = true to mark that a read operation is in progress.
• It invokes your custom _read() method to fetch more data.

Here’s a simplified snippet of the relevant logic:

// Inside Readable.read()
if (state.length < n && !state.reading) {
  state.reading = true;
  try {
    this._read(state.highWaterMark);
  } catch (err) {
    // Error handling logic
  }
}

(Note: state.highWaterMark is the default buffer size, but in your case, the 5-byte demand from head is what drives repeated calls, not the high water mark.)

2. The push() Method: Resetting and Checking Demand #

When you call rs.push() in your _read() method, the stream adds the data to its internal buffer, then:

• Sets state.reading = false to indicate the current read operation is complete.
• Calls maybeReadMore() to check if more data is needed to meet the consumer’s demand.

The maybeReadMore() function looks roughly like this:

function maybeReadMore(stream) {
  const state = stream._readableState;
  while (!state.reading && state.needReadable && state.length < state.highWaterMark) {
    stream.read(0); // Triggers another read cycle
  }
}

In your case, after each 1-byte push, state.length is still less than the 5-byte demand from head, so maybeReadMore() triggers another call to read(), which in turn calls your _read() method again. This loop continues until state.length meets or exceeds the consumer’s need.

3. Pipe and Backpressure: The Consumer Calls the Shots #

When you use rs.pipe(process.stdout), Node.js sets up a backpressure mechanism where the consumer (process.stdout, controlled by head) tells the producer (your Readable stream) how much data it can handle. As soon as head receives its 5 bytes, it closes the stream, which signals your Readable to stop calling _read().

Why Exactly 5 Calls? #

Each _read() call in your code only pushes 1 byte. Since head requests 5 bytes total, the stream needs to call _read() 5 times to accumulate enough data to satisfy the demand. If you modified your _read() to push all 5 bytes at once, it would only be called a single time!

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