stream.Duplex在Node.js流体系中是一个同时具备可读和可写能力的抽象类。它经常被误认为只是把Readable和Writable简单拼接在一起，但它的实现细节会影响数据流动和资源释放。一个典型的Duplex实例内部维护着两个独立缓冲区：可读端缓冲区由Readable管理，可写端缓冲区由Writable管理。二者在默认情况下互不干扰，数据不会自动从写端流向读端。net.Socket、tls.TLSSocket以及许多基于TCP连接的封装都属于Duplex流，写入到socket的数据会发送给远端，远端发来的数据则进入读缓冲区等待应用消费。

自定义Duplex的核心任务就是决定两件事：当读端需要数据时做什么，当写端拿到数据时做什么。前者通过_read实现，后者通过_write实现。下面从内部执行机制开始拆解。

一、_read与_write的触发时机和协作方式

当流内部读缓冲区低于highWaterMark，或者消费者调用read()时，Node.js会调用_read。在_read中，开发者可以调用this.push(chunk)把数据塞入读缓冲区。如果暂时没有数据，可以什么都不做，但要避免同步阻塞进程。异步数据到达后，依然可以在事件回调里调用push，无需等待_read被再次调用。push方法返回布尔值，为false表示读缓冲区已经达到阈值，底层数据源应当暂停生产。

可写端那边，上游调用write(chunk)时，数据先进入写缓冲区，随后Node.js调用_write(chunk, encoding, callback)。在_write内部必须把chunk写入真正的底层目标，例如设备、远端连接或内存队列。处理成功后调用callback()，表示可以继续处理下一块写缓冲数据；处理失败则调用callback(err)，错误会触发流的error事件并进入销毁流程。

可写端与可读端不自动连接，这是Duplex与Transform最大的差异。Duplex适合包装一个已存在的双向通道，而Transform适合对数据进行加工。例如你拿到一个WebSocket连接，它既能发送消息也能接收消息，用Duplex包装它时，_write调用底层发送函数，底层message事件调用push。写入的数据不会自动成为读端输出，两条通道的数据流由开发者显式控制。

二、实现一个自定义Duplex：内存队列与设备封装

先看一个最简单的类继承写法。它维护一个内存队列，可读端从队列中取数据，可写端把数据放入队列，读写两端各司其职。这个例子没有涉及真正的异步设备，但足够说明回调如何驱动数据流动。

const { Duplex } = require('stream');
class MemoryDuplex extends Duplex {
  constructor() {
    super({ allowHalfOpen: false });
    this.queue = [];
  }
  _read(size) {
    while (this.queue.length) {
      const chunk = this.queue.shift();
      if (!this.push(chunk)) {
        return;
      }
    }
  }
  _write(chunk, encoding, callback) {
    this.queue.push(chunk);
    callback();
  }
  _final(callback) {
    this.push(null);
    callback();
  }
}

在这个实现中，_write只负责把数据放入数组并立即回调，不直接操作读端。当消费者开始读取时，Node.js会触发_read，_read再尽可能把队列里的数据推入读缓冲区。如果push返回false，说明读缓冲区已经满了，_read直接返回，等下一次读端腾出空间后再继续取出队列数据。这样写端速度再快，也不会直接撑爆读缓冲区。

更贴近真实设备的是事件驱动写法。底层设备在数据到达时通过事件通知，我们在这个事件里尝试push。如果push返回false，需要暂停底层读取；等消费者消耗一部分数据后，_read会被再次调用，此时再恢复底层读取。下面的实现演示了如何平稳处理背压。

const { Duplex } = require('stream');
class DeviceDuplex extends Duplex {
  constructor(device) {
    super();
    this.device = device;
    const self = this;
    this.onDeviceData = function(chunk) {
      if (!self.push(chunk)) {
        self.device.pause();
      }
    };
    this.device.on('data', this.onDeviceData);
    this.device.on('end', function() {
      self.push(null);
    });
  }
  _read() {
    this.device.resume();
  }
  _write(chunk, encoding, callback) {
    this.device.write(chunk, encoding, callback);
  }
  _final(callback) {
    this.device.end(callback);
  }
  _destroy(err, callback) {
    this.device.removeListener('data', this.onDeviceData);
    super._destroy(err, callback);
  }
}

这个示例中，设备主动上报的数据通过事件回调进入push。一旦读缓冲区达到上限，push返回false，我们就暂停设备数据流。之后消费者继续调用read()，Node.js再次进入_read，在其中调用device.resume()恢复设备输出。写端则把外部写入的数据直接转发给设备，并在设备确认写入完成后才调用回调。

释放资源时，_destroy负责移除事件监听，并调用父类的_destroy完成清理。不要把这些逻辑放在_final里，因为_final只表示写端已经开始结束，不代表对象已经销毁。

三、Duplex与Transform的选择以及关闭语义

不少人会在Duplex的_write里直接调用this.push(chunk)，这样确实能让写入的数据从读端被读出来，看起来很像一个回显流。但这本质上是把写缓冲区的数据复制到读缓冲区，并没有经过任何转换，而且如果消费者读取速度跟不上写入速度，push会返回false。忽视这个返回值会破坏背压，读缓冲区可能无限增长。任何时候在事件回调或_write里调用push，都应该检查返回值并暂停数据源。

如果业务目标真的是让数据经过一定处理后从读端输出，应该使用stream.Transform。Transform继承自Duplex，它的缓冲区和数据流是串行连接的：_write接收数据后进入_transform(chunk, encoding, callback)，在_transform中处理后调用this.push(transformedChunk)，最后调用callback。Duplex则保留读写两端独立，适合同一个对象既处理输入又产生无关联输出的场景。例如你封装一个串口设备，电脑写进去的指令和串口主动上报的状态数据在协议上属于不同通道，就不应该让写入指令自动变成读端输出。

关闭语义也容易混淆。end()只表示写端没有更多数据要写入，它会等待写缓冲清空后触发_final。默认情况下，写端结束后读端仍然可以继续产出数据，这是allowHalfOpen为true时的行为。如果设置{ allowHalfOpen: false }，写端结束后会自动结束读端。无论哪种模式，当你确定底层数据已经读完时，应该显式调用push(null)结束读端，否则消费者会一直等待下一个chunk。在_final中调用push(null)是常见做法，但要注意_final只会在写端结束时执行，资源清理应放在_destroy中。

四、实战：封装子进程标准输入输出为一个双向会话流

子进程的stdin只可写，stdout只可读。如果想把一个命令行工具包装成可写可读的会话对象，Duplex非常合适。下面是一个可运行的思路：创建子进程，监听stdout数据并放入Duplex读缓冲，写操作直接转发给stdin，同时处理背压和进程退出。

const { Duplex } = require('stream');
const { spawn } = require('child_process');
function createProcessDuplex(command, args) {
  const child = spawn(command, args || [], {
    stdio: ['pipe', 'pipe', 'pipe']
  });
  const duplex = new Duplex({
    allowHalfOpen: false,
    read() {
      child.stdout.resume();
    },
    write(chunk, encoding, callback) {
      child.stdin.write(chunk, encoding, callback);
    },
    final(callback) {
      child.stdin.end(callback);
    },
    destroy(err, callback) {
      if (!child.killed) {
        child.kill();
      }
      callback(err);
    }
  });
  child.stdout.on('data', function(chunk) {
    if (!duplex.push(chunk)) {
      child.stdout.pause();
    }
  });
  child.stdout.on('end', function() {
    duplex.push(null);
  });
  child.stderr.pipe(process.stderr);
  child.on('error', function(err) {
    duplex.destroy(err);
  });
  child.on('close', function() {
    duplex.destroy();
  });
  return duplex;
}

当duplex.push(chunk)返回false时，child.stdout.pause()会暂停读取子进程输出。当消费者从duplex中读走数据后，Node.js会再次调用read()，在其中执行child.stdout.resume()恢复输出，这样就形成了闭环背压。子进程的stderr直接管道到父进程的stderr，避免它与主要数据流混杂在一起。

如果子进程本身可能产生大量错误输出，建议将stderr单独收集或写入日志文件，不要直接在数据回调中做耗时操作。另一个典型用法是将两个Duplex连接起来做代理：A端写入时转发给B端可写端，B端读到的数据转发给A端写端。在此基础上处理EOF、半关闭和错误传播，可以实现TCP代理或协议网关。Duplex的优势正是把两端的状态收敛到一个对象，减少中间管道拼接的复杂度。

五、调试与性能建议

使用Duplex时，最容易出现的问题是没有监听error事件。当_write回调传递错误，或者底层资源触发错误时，流会进入销毁流程。若没有监听error事件，Node.js会抛出未捕获异常。建议在返回流或启动操作前统一添加stream.on('error', handler)，并把错误日志与资源清理放在一起。

性能方面，避免在_read中阻塞等待数据，也不要反复压入过小的chunk。可以适当在设备侧合并数据块后再调用push；写端也要避免把一个很大的Buffer一次性写入后不检查回调时延。利用好的背压机制能让CPU和内存都保持稳健。最后，用stream.Duplex前先确认底层是否已经是Duplex，例如net.Socket本身就能满足双向读写，重复包装只会增加调用层级。