class MyDuplex extends Duplex{ constructor(){ super(); this.source = []; } _read(){ if (!this.source.length) { this.push(null); } else { this.push(this.source.shift()); } } _write(chunk,encoding,cb){ this.source.push(chunk); cb(); } } const myDuplex = new MyDuplex(); myDuplex.on('finish',()=>{ process.stdout.write('write done.') }); myDuplex.on('end',()=>{ process.stdout.write('read done.') }); myDuplex.write('\na\n'); myDuplex.write('c\n'); myDuplex.end('b\n'); myDuplex.pipe(process.stdout);
上面的代码实现了_read()方法，可作为可读流来使用，同时实现了_write()方法，又可作为可写流来使用。
Transform流实现
实现Transform流，需要继承Transform类，并实现_transform接口，如下所示：
class MyTransform extends Transform{ constructor(){ super(); } _transform(chunk, encoding, callback){ chunk = (chunk+'').toUpperCase(); callback(null,chunk); } } const myTransform = new MyTransform(); myTransform.write('hello world!'); myTransform.end(); myTransform.pipe(process.stdout);
上面代码中的_transform()方法，其第一个参数，要么为error，要么为null，第二个参数将被自动转发给readable.push()方法，因此该方法也可以使用如下写法：
_transform(chunk, encoding, callback){ chunk = (chunk+'').toUpperCase() this.push(chunk) callback(); }
Object Mode流实现
我们知道流中的数据默认都是Buffer类型，可读流的数据进入流中便被转换成buffer，然后被消耗，可写流写入数据时，底层调用也将其转化为buffer。但将构造函数的objectMode选择设置为true，便可产生原样的数据，如下所示：
const rs = Readable(); rs.push('a'); rs.push('b'); rs.push(null); rs.on('data',(chunk)=>{console.log(chunk);});//<Buffer 61>与<Buffer 62> const rs1 = Readable({objectMode:!0}); rs1.push('a'); rs1.push('b'); rs1.push(null); rs1.on('data',(chunk)=>{console.log(chunk);});//a与b
下面利用Transform流实现一个简单的CSS压缩工具，如下所示：
function minify(src,dest){ const transform = new Transform({ transform(chunk,encoding,cb){ cb(null,(chunk.toString()).replace(/[\s\r\n\t]/g,'')); } }); fs.createReadStream(src,{encoding:'utf8'}).pipe(transform).pipe(fs.createWriteStream(dest)); } minify('./reset.css','./reset.min.css');