NodeJS Stream 五：雙工流

雙工流就是同時實現了 Readable 和 Writable 的流，便可以做爲上游生產數據，又能夠做爲下游消費數據，這樣能夠處於數據流動管道的中間部分，即

rs.pipe(rws1).pipe(rws2).pipe(rws3).pipe(ws);

在 NodeJS 中雙工流經常使用的有兩種

1. Duplex
2. Transform

Duplex

實現 Duplex

和 Readable、Writable 實現方法相似，實現 Duplex 流很是簡單，但 Duplex 同時實現了 Readable 和 Writable， NodeJS 不支持多繼承，因此咱們須要繼承 Duplex 類

1. 繼承 Duplex 類
2. 實現 _read() 方法
3. 實現 _write() 方法

相信你們對 read()、write() 方法的實現不會陌生，由於和 Readable、Writable 徹底同樣。

const Duplex = require('stream').Duplex;

const myDuplex = new Duplex({
  read(size) {
    // ...
  },
  write(chunk, encoding, callback) {
    // ...
  }
});

構造函數參數

Duplex 實例內同時包含可讀流和可寫流，在實例化 Duplex 類的時候能夠傳遞幾個參數

• readableObjectMode : 可讀流是否設置爲 ObjectMode，默認 false
• writableObjectMode : 可寫流是否設置爲 ObjectMode，默認 false
• allowHalfOpen : 默認 true， 設置成 false 的話，當寫入端結束的時，流會自動的結束讀取端，反之亦然。

小例子

瞭解了 Readable 和 Writable 以後看 Duplex 很是簡單，直接用一個官網的例子

const Duplex = require('stream').Duplex;
const kSource = Symbol('source');

class MyDuplex extends Duplex {
  constructor(source, options) {
    super(options);
    this[kSource] = source;
  }

  _write(chunk, encoding, callback) {
    // The underlying source only deals with strings
    if (Buffer.isBuffer(chunk))
      chunk = chunk.toString();
    this[kSource].writeSomeData(chunk);
    callback();
  }

  _read(size) {
    this[kSource].fetchSomeData(size, (data, encoding) => {
      this.push(Buffer.from(data, encoding));
    });
  }
}

固然這是不能執行的僞代碼，可是 Duplex 的做用可見一斑，進能夠生產數據，又能夠消費數據，因此才能夠處於數據流動管道的中間環節，常見的 Duplex 流有

• Tcp Scoket
• Zlib
• Crypto

Transform

Transform 一樣是雙工流，看起來和 Duplex 重複了，但二者有一個重要的區別：Duplex 雖然同事具有可讀流和可寫流，但二者是相對獨立的；Transform 的可讀流的數據會通過必定的處理過程自動進入可寫流。

雖然會從可讀流進入可寫流，但並不意味這二者的數據量相同，上面說的必定的處理邏輯會決定若是 tranform 可讀流，而後放入可寫流，transform 原義即爲轉變，很貼切的描述了 Transform 流做用。

咱們最多見的壓縮、解壓縮用的 zlib 即爲 Transform 流，壓縮、解壓先後的數據量明顯不一樣，兒流的做用就是輸入一個 zip 包，輸入一個解壓文件或反過來。咱們平時用的大部分雙工流都是 Transform。

實現 Tranform

Tranform 類內部繼承了 Duplex 並實現了 writable.write() 和 readable._read() 方法，咱們想自定義一個 Transform 流，只須要

1. 繼承 Transform 類
2. 實現 _transform() 方法
3. 實現 _flush() 方法（能夠不實現）

_transform(chunk, encoding, callback) 方法用來接收數據，併產生輸出，參數咱們已經很熟悉了，和 Writable 同樣， chunk 默認是 Buffer，除非 decodeStrings 被設置爲 false。

在 _transform() 方法內部能夠調用 this.push(data) 生產數據，交給可寫流，也能夠不調用，意味着輸入不會產生輸出。

當數據處理完了必須調用 callback(err, data) ，第一個參數用於傳遞錯誤信息，第二個參數能夠省略，若是被傳入了，效果和 this.push(data) 同樣

transform.prototype._transform = function (data, encoding, callback) {
  this.push(data);
  callback();
};

transform.prototype._transform = function (data, encoding, callback) {
  callback(null, data);
};

有些時候，transform 操做可能須要在流的最後多寫入可寫流一些數據。例如， Zlib流會存儲一些內部狀態，以便優化壓縮輸出。在這種狀況下，可使用_flush()方法，它會在全部寫入數據被消費、觸發 'end'以前被調用。

Transform 事件

Transform 流有兩個經常使用的事件

1. 來自 Writable 的 finish
2. 來自 Readable 的 end

當調用 transform.end() 而且數據被 _transform() 處理完後會觸發 finish，調用_flush後，全部的數據輸出完畢，觸發end事件。

對比

瞭解了 Readable 和 Writable 以後，理解雙工流十分天然，但二者的區別會讓一些初學者困惑，簡單的區分：Duplex 的可讀流和可寫流之間並無直接關係，Transform 中可讀流的數據會通過處理後自動放入可寫流中。

看兩個簡單的例子就能直觀瞭解到 Duplex 和 Transform 的區別

TCP socket

net 模塊能夠用來建立 socket，socket 在 NodeJS 中是一個典型的 Duplex，看一個 TCP 客戶端的例子

var net = require('net');

//建立客戶端
var client = net.connect({port: 1234}, function() {
    console.log('已鏈接到服務器');
    client.write('Hi!');
});

//data事件監聽。收到數據後，斷開鏈接
client.on('data', function(data) {
    console.log(data.toString());
    client.end();
});

//end事件監聽，斷開鏈接時會被觸發
client.on('end', function() {
    console.log('已與服務器斷開鏈接');
});

能夠看到 client 就是一個 Duplex，可寫流用於向服務器發送消息，可讀流用於接受服務器消息，兩個流內的數據並無直接的關係。

gulp

gulp 很是擅長處理代碼本地構建流程，看一段官網的示例代碼

gulp.src('client/templates/*.jade')
  .pipe(jade())
  .pipe(minify())
  .pipe(gulp.dest('build/minified_templates'));

其中 jada() 和 minify() 就是典型的 Transform，處理流程大概是

.jade 模板文件 -> jada() -> html 文件 -> minify -> 壓縮後的 html

能夠看出來，jade() 和 minify() 都是對輸入數據作了些特殊處理，而後交給了輸出數據。

這樣簡單的對比就能看出 Duplex 和 Transform 的區別，在平時實用的時候，當一個流同事面向生產者和消費者服務的時候咱們會選擇 Duplex，當只是對數據作一些轉換工做的時候咱們便會選擇使用 Tranform。