如何在Node.js原生Stream API中实现Web Stream的pull等效功能？

Node.js原生的stream.Readable类中，_read方法就对应Web Stream API里的pull函数——当流的消费者需要更多数据时，Node.js会自动调用_read方法来触发数据的获取和推送。

实现步骤 #

1. 继承stream.Readable类，重写_read方法：

   • 在_read中发起HTTP请求获取数据（Node.js 18+内置fetch，低版本可使用node-fetch包）
   • 拿到数据后，用this.push()将数据推入流中
   • 如果没有更多数据或请求失败，调用this.push(null)结束流

2. 读取流时，无论是用事件监听还是getReader()，流都会自动触发_read来拉取新数据，和你之前Web Stream的行为一致。

完整代码示例 #

const { Readable } = require('stream');
// 若Node.js版本低于18，需先安装node-fetch：npm install node-fetch，然后引入
// const fetch = require('node-fetch');

class RestServiceStream extends Readable {
  constructor(options) {
    // 设置objectMode为true，因为我们要推送JSON对象而非Buffer
    super({ ...options, objectMode: true });
  }

  _read() {
    // 发起请求获取数据
    fetch('http://restservice/to/fetch/items')
      .then(response => {
        if (!response.ok) {
          throw new Error(`HTTP error! status: ${response.status}`);
        }
        return response.json();
      })
      .then(item => {
        // 将数据推入流
        this.push(item);
        // 当消费者读取完当前数据，_read会自动再次被调用，实现持续拉取
        // 若要停止拉取（比如无更多数据），调用this.push(null)即可
      })
      .catch(err => {
        console.error('获取数据失败:', err);
        this.push(null); // 出错后结束流
      });
  }
}

// 使用流
const stream = new RestServiceStream();

// 方式1：用事件监听读取
stream.on('data', (item) => {
  console.log('收到数据:', item);
});

stream.on('end', () => {
  console.log('流已结束');
});

// 方式2：用getReader()（Node.js 16+支持Web Stream兼容的API）
// const reader = stream.getReader();
// function readNext() {
//   reader.read().then(({ done, value }) => {
//     if (done) {
//       console.log('流已结束');
//       return;
//     }
//     console.log('收到数据:', value);
//     readNext();
//   });
// }
// readNext();

关键说明 #

• objectMode: true：默认Node.js流处理Buffer/字符串，我们要推送JSON对象，必须开启这个选项。
• _read的自动触发：每当流的内部缓冲区有空闲空间（消费者读取了数据），Node.js就会自动调用_read方法，实现和Web Stream中pull一致的“按需拉取”逻辑。
• 结束流：当没有更多数据或发生错误时，调用this.push(null)，流会触发end事件。

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