如何在Node.js的Readable与Writable流之间通过自定义逻辑实现管道传输？

我有两个流：传入的readable流和传出的writable流。初始配置中直接将它们管道连接就能正常运行，但我的场景需要对传输的数据进行处理，于是实现了如下逻辑：

readable.on("data", (data) => {
  const modifiedData = modify(data);
  writable.write(modifiedData)
})
readable.on("finish", () => writable.end())

这种方式下两个流彼此不感知，还得手动结束writable流。想知道怎么创建中间逻辑，实现readable.pipe(myLogic).pipe(writable)的调用方式？

你需要创建转换流（Transform Stream）——它是Node.js流体系中专门用于数据转换的双工流，同时具备可读和可写接口，完美适配这种管道串联的场景。

实现自定义转换流（类写法） #

基于stream模块的Transform类继承实现，适合复杂逻辑：

const { Transform } = require('stream');

class MyLogicTransform extends Transform {
  constructor(options = {}) {
    super(options);
  }

  // 核心处理方法：接收上游数据，处理后推送给下游
  _transform(chunk, encoding, callback) {
    try {
      // 替换成你的数据修改逻辑
      const modifiedData = modify(chunk);
      this.push(modifiedData);
      callback();
    } catch (err) {
      // 传递错误，终止流处理
      callback(err);
    }
  }

  // 可选：流结束前的收尾操作（比如处理剩余数据）
  _flush(callback) {
    // 如有收尾逻辑，写在这里
    callback();
  }
}

// 创建转换流实例
const myLogic = new MyLogicTransform();

// 管道串联即可，无需手动结束下游流
readable.pipe(myLogic).pipe(writable);

简化写法（直接实例化Transform） #

如果逻辑简单，无需封装成类，直接用构造函数创建：

const { Transform } = require('stream');

const myLogic = new Transform({
  transform(chunk, encoding, callback) {
    const modifiedData = modify(chunk);
    this.push(modifiedData);
    callback();
  },
  // 可选flush方法
  flush(callback) {
    callback();
  }
});

readable.pipe(myLogic).pipe(writable);

优势说明 #

• 自动处理流的生命周期：上游流结束时，转换流会自动触发下游流的结束，无需手动调用writable.end()。
• 内置背压处理：避免手动监听data事件可能导致的内存溢出问题。
• 符合Node.js流的标准规范，和原生流的行为完全兼容。

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