穩(wěn)定性: 2 - 不穩(wěn)定

流用于處理Node.js中的流數據的抽象接口，在Node里被不同的對象實現。例如，對HTTP服務器的請求是流，process.stdout 是流。

流是可讀的，可寫的，或者是可讀寫的，所有的流是EventEmitter的實例。

Node.js訪問流模塊的方法如下所示：

const stream = require('stream');

你可以通過require('stream')加載Stream基類。其中包括了Readable流、Writable流、Duplex流和Transform流的基類。

本文將分為3個部分進行介紹。

第一個部分解釋了在你的程序中使用流時候需要了解的內容。如果你不用實現流式API，可以只看這個部分。

如果你想實現你自己的流，第二個部分解釋了這部分API。這些API讓你的實現更加簡單。

第三個部分深入的解釋了流是如何工作的，包括一些內部機制和函數，這些內容不要改動，除非你明確知道你要做什么。

面向流消費者的API

流可以是可讀（Readable），可寫（Writable），或者是可讀可寫的（Duplex，雙工）。

所有的流都是事件分發(fā)器（EventEmitters），但是也有自己的方法和屬性，這取決于他它們是可讀（Readable），可寫（Writable），或者兼具兩者（Duplex，雙工）的。

如果流是可讀寫的，則它實現了下面的所有方法和事件。因此，這個部分API完全闡述了Duplex或Transform流，即便他們的實現有所不同。

沒有必要為了消費流而在你的程序里實現流的接口。如果你正在你的程序里實現流接口，請同時參考下面的流實現程序API。

基本所有的Node程序，無論多簡單，都會使用到流。以下是一個使用流的例子：

javascript
var http = require('http');

var server = http.createServer(function (req, res) {
  // req is an http.IncomingMessage, which is 可讀流（Readable stream）
  // res is an http.ServerResponse, which is a Writable Stream

  var body = '';
  // we want to get the data as utf8 strings
  // If you don't set an encoding, then you'll get Buffer objects
  req.setEncoding('utf8');

  // 可讀流（Readable stream） emit 'data' 事件 once a 監(jiān)聽器（listener） is added
  req.on('data', function (chunk) {
    body += chunk;
  });

  // the end 事件 tells you that you have entire body
  req.on('end', function () {
    try {
      var data = JSON.parse(body);
    } catch (er) {
      // uh oh!  bad json!
      res.statusCode = 400;
      return res.end('error: ' + er.message);
    }

    // write back something interesting to the user:
    res.write(typeof data);
    res.end();
  });
});

server.listen(1337);

// $ curl localhost:1337 -d '{}'
// object
// $ curl localhost:1337 -d '"foo"'
// string
// $ curl localhost:1337 -d 'not json'
// error: Unexpected token o

類： stream.Readable

可讀流（Readable stream）接口是對你正在讀取的數據的來源的抽象。換句話說，數據來來自

可讀流（Readable stream）不會分發(fā)數據，直到你表明準備就緒。

可讀流（Readable stream） 有2種模式: 流動模式（flowing mode）和暫停模式（paused mode）。流動模式（flowing mode）時，盡快的從底層系統(tǒng)讀取數據并提供給你的程序。暫停模式（paused mode）時，你必須明確的調用stream.read()來讀取數據。暫停模式（paused mode）是默認模式。

注意: 如果沒有綁定數據處理函數，并且沒有pipe()目標，流會切換到流動模式（flowing mode），并且數據會丟失。

可以通過下面幾個方法，將流切換到流動模式（flowing mode）。

• 添加一個['data' 事件][]事件處理器來監(jiān)聽數據.
• 調用resume()方法來明確的開啟數據流。
• 調用pipe()方法來發(fā)送數據給Writable.

可以通過以下方法來切換到暫停模式（paused mode）:

• 如果沒有“導流（pipe）”目標，調用pause()方法.
• 如果有“導流（pipe）”目標，移除所有的['data'事件][]處理函數，調用unpipe()方法移除所有的“導流（pipe）”目標。

注意：為了向后兼容考慮， 移除'data'事件監(jiān)聽器并不會自動暫停流。同樣的，當有導流目標時，調用pause()并不能保證流在那些目標排空后，請求更多數據時保持暫停狀態(tài)。

可讀流（Readable stream）例子包括:

• http responses, on the client
• http requests, on the server
• fs read streams
• zlib streams
• crypto streams
• tcp sockets
• child process stdout and stderr
• process.stdin

事件: 'readable'

當一個數據塊可以從流中讀出，將會觸發(fā)'readable'事件.`

某些情況下, 如果沒有準備好，監(jiān)聽一個'readable'事件將會導致一些數據從底層系統(tǒng)讀取到內部緩存。

javascript
var readble = getReadableStreamSomehow();
readable.on('readable', function() {
  // there is some data to read now
});

一旦內部緩存排空，一旦有更多數據將會再次觸發(fā)readable事件。

事件: 'data'

• chunk {Buffer | String} 數據塊

綁定一個data事件的監(jiān)聽器（listener）到一個未明確暫停的流，會將流切換到流動模式。數據會盡額能的傳遞。

如果你像盡快的從流中獲取數據，以下的方法是最快的：

javascript
var readable = getReadableStreamSomehow();
readable.on('data', function(chunk) {
  console.log('got %d bytes of data', chunk.length);
});

事件: 'end'

如果沒有更多的可讀數據，將會觸發(fā)這個事件。

注意：只有數據已經被完全消費，end事件才會觸發(fā)?？梢酝ㄟ^切換到流動模式（flowing mode）來實現，或者通過調用重復調用read()獲取數據，直到結束。

javascript
    var readable = getReadableStreamSomehow();
    readable.on('data', function(chunk) {
        console.log('got %d bytes of data', chunk.length);
    });
    readable.on('end', function() {
        console.log('there will be no more data.');
    });  

事件: 'close'

當底層資源（例如源頭的文件描述符）關閉時觸發(fā)。并不是所有流都會觸發(fā)這個事件。

事件: 'error'

• {Error Object}

當接收數據時發(fā)生錯誤觸發(fā)。

readable.read([size])

• size {Number} 可選參數， 需要讀入的數據量
• 返回 {String | Buffer | null}

read()方法從內部緩存中拉取數據。如果沒有可用數據，將會返回null

如果傳了size參數，將會返回相當字節(jié)的數據。如果size不可用，將會返回null。

如果你沒有指定size參數。將會返回內部緩存的所有數據。

這個方法僅能再暫停模式（paused mode）里調用。 流動模式（flowing mode）下這個方法會被自動調用直到內存緩存排空。

javascript
var readable = getReadableStreamSomehow();
readable.on('readable', function() {
  var chunk;
  while (null !== (chunk = readable.read())) {
    console.log('got %d bytes of data', chunk.length);
  }
});

如果這個方法返回一個數據塊, 它同時也會觸發(fā)['data'事件][].

readable.setEncoding(encoding)

• encoding {String} 要使用的編碼.
• 返回:this

調用此函數會使得流返回指定編碼的字符串，而不是Buffer對象。例如，如果你調用readable.setEncoding('utf8')，輸出數據將會是UTF-8編碼，并且返回字符串。如果你調用readable.setEncoding('hex')，將會返回2進制編碼的數據。

該方法能正確處理多字節(jié)字符。如果不想這么做，僅簡單的直接拉取緩存并調buf.toString(encoding)，可能會導致字節(jié)錯位。因此，如果你想以字符串讀取數據，請使用下述的方法：

javascript
var readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', function(chunk) {
  assert.equal(typeof chunk, 'string');
  console.log('got %d characters of string data', chunk.length);
});

readable.resume()

• 返回:this

這個方法讓可讀流（Readable stream）繼續(xù)觸發(fā)data事件.

這個方法會將流切換到流動模式（flowing mode）。 如果你不想從流中消費數據，而想得到end事件，可以調用readable.resume()來打開數據流，如下所示：

javascript
var readable = getReadableStreamSomehow();
readable.resume();
readable.on('end', function(chunk) {
  console.log('got to the end, but did not read anything');
});

readable.pause()

• 返回:this

這個方法會使得流動模式（flowing mode）的流停止觸發(fā)data事件，切換到流動模式（flowing mode）。并讓后續(xù)可用數據留在內部緩沖區(qū)中。

javascript
var readable = getReadableStreamSomehow();
readable.on('data', function(chunk) {
  console.log('got %d bytes of data', chunk.length);
  readable.pause();
  console.log('there will be no more data for 1 second');
  setTimeout(function() {
    console.log('now data will start flowing again');
    readable.resume();
  }, 1000);
});

readable.isPaused()

• 返回:Boolean

這個方法返回readable是否被客戶端代碼明確的暫停（調用readable.pause()）。

var readable = new stream.Readable

readable.isPaused() // === false
readable.pause()
readable.isPaused() // === true
readable.resume()
readable.isPaused() // === false

readable.pipe(destination[, options])

• destination {Writable Stream} 寫入數據的目標
• options {Object} 導流（pipe）選項
  • end {Boolean} 讀取到結束符時，結束寫入者。默認 = true

這個方法從可讀流（Readable stream）拉取所有數據，并將數據寫入到提供的目標中。自動管理流量，這樣目標不會快速的可讀流（Readable stream）淹沒。

可以導流到多個目標。

javascript
var readable = getReadableStreamSomehow();
var writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt'
readable.pipe(writable);

這個函數返回目標流, 因此你可以建立導流鏈：

javascript
var r = fs.createReadStream('file.txt');
var z = zlib.createGzip();
var w = fs.createWriteStream('file.txt.gz');
r.pipe(z).pipe(w);

例如：模擬Unix的cat命令：

javascript
process.stdin.pipe(process.stdout);

默認情況下，當源數據流觸發(fā)end的時候調用end()，所以destination不可再寫。傳{ end:false }作為options，可以保持目標流打開狀態(tài)。

這會讓writer保持打開狀態(tài)，可以在最后寫入"Goodbye"：

javascript
reader.pipe(writer, { end: false });
reader.on('end', function() {
  writer.end('Goodbye\n');
});

注意：process.stderr和process.stdout直到進程結束才會關閉，無論是否指定它們。

readable.unpipe([destination])

• destination {Writable Stream} 可選，指定解除導流的流

這個方法會解除之前調用pipe() 設置的鉤子（pipe()）。

如果沒有指定destination，則所有的導流（pipe）都會被移除。

如果指定了destination，但是沒有建立如果沒有指定destination，則什么事情都不會發(fā)生。

javascript
var readable = getReadableStreamSomehow();
var writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt',
// but only for the first second
readable.pipe(writable);
setTimeout(function() {
  console.log('stop writing to file.txt');
  readable.unpipe(writable);
  console.log('manually close the file stream');
  writable.end();
}, 1000);

readable.unshift(chunk)

• chunk {Buffer | String} 數據塊插入到讀隊列中

這個方法很有用，當一個流正被一個解析器消費，解析器可能需要將某些剛拉取出的數據“逆消費”，返回到原來的源，以便流能將它傳遞給其它消費者。

如果你在程序中必須經常調用stream.unshift(chunk) ，那你可以考慮實現Transform來替換（參見下文API for Stream Implementors）。

javascript
// Pull off a header delimited by \n\n
// use unshift() if we get too much
// Call the callback with (error, header, stream)
var StringDecoder = require('string_decoder').StringDecoder;
function parseHeader(stream, callback) {
  stream.on('error', callback);
  stream.on('readable', onReadable);
  var decoder = new StringDecoder('utf8');
  var header = '';
  function onReadable() {
    var chunk;
    while (null !== (chunk = stream.read())) {
      var str = decoder.write(chunk);
      if (str.match(/\n\n/)) {
        // found the header boundary
        var split = str.split(/\n\n/);
        header += split.shift();
        var remaining = split.join('\n\n');
        var buf = new Buffer(remaining, 'utf8');
        if (buf.length)
          stream.unshift(buf);
        stream.removeListener('error', callback);
        stream.removeListener('readable', onReadable);
        // now the body of the message can be read from the stream.
        callback(null, header, stream);
      } else {
        // still reading the header.
        header += str;
      }
    }
  }
}

readable.wrap(stream)

• stream {Stream} 一個舊式的可讀流（Readable stream）

v0.10版本之前的Node流并未實現現在所有流的API（更多信息詳見下文“兼容性”部分）。

如果你使用的是舊的Node庫，它觸發(fā)'data'事件，并擁有僅做查詢用的pause()方法，那么你能使用wrap()方法來創(chuàng)建一個Readable流來使用舊版本的流，作為數據源。

你應該很少需要用到這個函數，但它會留下方便和舊版本的Node程序和庫交互。

例如:

javascript
var OldReader = require('./old-api-module.js').OldReader;
var oreader = new OldReader;
var Readable = require('stream').Readable;
var myReader = new Readable().wrap(oreader);

myReader.on('readable', function() {
  myReader.read(); // etc.
});

類： stream.Writable

可寫流（Writable stream ）接口是你正把數據寫到一個目標的抽象。

可寫流（Writable stream ）的例子包括:

• http requests, on the client
• http responses, on the server
• fs write streams
• zlib streams
• crypto streams
• tcp sockets
• child process stdin
• process.stdout, process.stderr

writable.write(chunk[, encoding][, callback])

• chunk {String | Buffer} 準備寫的數據
• encoding {String} 編碼方式（如果chunk 是字符串）
• callback {Function} 數據塊寫入后的回調
• 返回: {Boolean} 如果數據已被全部處理返回true

這個方法向底層系統(tǒng)寫入數據，并在數據處理完畢后調用所給的回調。

返回值表示你是否應該繼續(xù)立即寫入。如果數據要緩存在內部，將會返回false。否則返回true。

返回值僅供參考。即使返回false，你也可能繼續(xù)寫。但是寫會緩存在內存里，所以不要做的太過分。最好的辦法是等待drain事件后，再寫入數據。

事件: 'drain'

如果調用writable.write(chunk)返回 false，drain事件會告訴你什么時候將更多的數據寫入到流中。

javascript
// Write the data to the supplied 可寫流（Writable stream ） 1MM times.
// Be attentive to back-pressure.
function writeOneMillionTimes(writer, data, encoding, callback) {
  var i = 1000000;
  write();
  function write() {
    var ok = true;
    do {
      i -= 1;
      if (i === 0) {
        // last time!
        writer.write(data, encoding, callback);
      } else {
        // see if we should continue, or wait
        // don't pass the callback, because we're not done yet.
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // had to stop early!
      // write some more once it drains
      writer.once('drain', write);
    }
  }
}

writable.cork()

強制緩存所有寫入。

調用.uncork()或.end()后，會把緩存數據寫入。

writable.uncork()

寫入所有.cork()調用之后緩存的數據。

writable.setDefaultEncoding(encoding)

• encoding {String} 新的默認編碼
• 返回:Boolean

給寫數據流設置默認編碼方式，如編碼有效，則返回true ，否則返回false。

writable.end([chunk][, encoding][, callback])

• chunk {String | Buffer} 可選，要寫入的數據
• encoding {String} 編碼方式（如果chunk是字符串）
• callback {Function} 可選， stream結束時的回調函數

當沒有更多的數據寫入的時候調用這個方法。如果給出，回調會被用作finish事件的監(jiān)聽器。

調用end()后調用write()會產生錯誤。

javascript
// write 'hello, ' and then end with 'world!'
var file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// writing more now is not allowed!

事件: 'finish'

調用end()方法后，并且所有的數據已經寫入到底層系統(tǒng)，將會觸發(fā)這個事件。

javascript
var writer = getWritableStreamSomehow();
for (var i = 0; i < 100; i ++) {
  writer.write('hello, #' + i + '!\n');
}
writer.end('this is the end\n');
writer.on('finish', function() {
  console.error('all writes are now complete.');
});

事件: 'pipe'

• src {Readable Stream} 是導流（pipe）到可寫流的源流。

無論何時在可寫流（Writable stream ）上調用pipe()方法，都會觸發(fā)'pipe'事件，添加這個流到目標。

javascript
var writer = getWritableStreamSomehow();
var reader = getReadableStreamSomehow();
writer.on('pipe', function(src) {
  console.error('something is piping into the writer');
  assert.equal(src, reader);
});
reader.pipe(writer);

事件: 'unpipe'

• src {Readable Stream}未寫入此可寫的源流。

無論何時在可寫流（Writable stream ）上調用unpipe()方法，都會觸發(fā)'unpipe'事件，將這個流從目標上移除。

javascript
var writer = getWritableStreamSomehow();
var reader = getReadableStreamSomehow();
writer.on('unpipe', function(src) {
  console.error('something has stopped piping into the writer');
  assert.equal(src, reader);
});
reader.pipe(writer);
reader.unpipe(writer);

事件: 'error'

• {Error object}

寫或導流（pipe）數據時，如果有錯誤會觸發(fā)。

類： stream.Duplex

雙工流（Duplex streams）是同時實現了Readable和Writable 接口。用法詳見下文。

雙工流（Duplex streams） 的例子包括:

• tcp sockets
• zlib streams
• crypto streams

類： stream.Transform

轉換流（Transform streams）是雙工Duplex流，它的輸出是從輸入計算得來。 它實現了Readable和Writable接口. 用法詳見下文.

轉換流（Transform streams）的例子包括:

• zlib streams
• crypto streams

流實現程序API

無論實現什么形式的流，模式都是一樣的:

1. 在你的子類中擴展適合的父類。 (util.inherits方法很有幫助)
2. 在你的構造函數中調用父類的構造函數，以確保內部的機制初始化正確。
3. 實現一個或多個方法，如下所列。

所擴展的類和要實現的方法取決于你要編寫的流類。

在你的代碼里，千萬不要調用流實現程序API里的方法。否則可能會引起消費流的程序副作用。

類： stream.Readable

stream.Readable是一個可被擴充的、實現了底層_read(size)方法的抽象類。

參照之前的流實現程序API查看如何在你的程序里消費流。以下內容解釋了在你的程序里如何實現可讀流（Readable stream）。

Example: 計數流

這是可讀流（Readable stream）的基礎例子，它將從1至1,000,000遞增地觸發(fā)數字，然后結束：

javascript
var Readable = require('stream').Readable;
var util = require('util');
util.inherits(Counter, Readable);

function Counter(opt) {
  Readable.call(this, opt);
  this._max = 1000000;
  this._index = 1;
}

Counter.prototype._read = function() {
  var i = this._index++;
  if (i > this._max)
    this.push(null);
  else {
    var str = '' + i;
    var buf = new Buffer(str, 'ascii');
    this.push(buf);
  }
};

Example: 簡單協(xié)議 v1 (初始版)

和之前描述的parseHeader函數類似，但它被實現為自定義流。注意這個實現不會將輸入數據轉換為字符串。

實際上，更好的辦法是將他實現為Transform流，使用下面的實現方法會更好：

javascript
// A parser for a simple data protocol.
// "header" is a JSON object, followed by 2 \n characters, and
// then a message body.
//
// 注意: This can be done more simply as a Transform stream!
// Using Readable directly for this is sub-optimal.  See the
// alternative example below under Transform section.

var Readable = require('stream').Readable;
var util = require('util');

util.inherits(SimpleProtocol, Readable);

  function SimpleProtocol(source, options) {
  if (!(this instanceof SimpleProtocol))
    return new SimpleProtocol(source, options);

  Readable.call(this, options;

  this._inBody = false;
  this._sawFirstCr = false;

  // source is 可讀流（Readable stream）, such as a socket or file
  this._source = source;

  var self = this;
  source.on('end', function() {
    self.push(null);
  });

  // give it a kick whenever the source is readable
  // read(0) will not consume any bytes
  source.on('readable', function() {
    self.read(0);
  });

  this._rawHeader = [];
  this.header = null;
}

SimpleProtocol.prototype._read = function(n) {
  if (!this._inBody) {
    var chunk = this._source.read();

    // if the source doesn't have data, we don't have data yet.
    if (chunk === null)
      return this.push('');

    // check if the chunk has a \n\n
    var split = -1;
    for (var i = 0; i < chunk.length; i++) {
      if (chunk[i] === 10) { // '\n'
        if (this._sawFirstCr) {
          split = i;
          break;
        } else {
          this._sawFirstCr = true;
        }
      } else {
        this._sawFirstCr = false;
      }
    }

    if (split === -1) {
      // still waiting for the \n\n
      // stash the chunk, and try again.
      this._rawHeader.push(chunk);
      this.push('');
    } else {
      this._inBody = true;
      var h = chunk.slice(0, split);
      this._rawHeader.push(h);
      var header = Buffer.concat(this._rawHeader).toString();
      try {
        this.header = JSON.parse(header);
      } catch (er) {
        this.emit('error', new Error('invalid simple protocol data'));
        return;
      }
      // now, because we got some extra data, unshift the rest
      // back into the 讀取隊列 so that our consumer will see it.
      var b = chunk.slice(split);
      this.unshift(b);

      // and let them know that we are done parsing the header.
      this.emit('header', this.header);
    }
  } else {
    // from there on, just provide the data to our consumer.
    // careful not to push(null), since that would indicate EOF.
    var chunk = this._source.read();
    if (chunk) this.push(chunk);
  }
};

// Usage:
// var parser = new SimpleProtocol(source);
// Now parser is 可讀流（Readable stream） that will emit 'header'
// with the parsed header data.

new stream.Readable([options])

• options{Object}
  • highWaterMark{Number} 停止從底層資源讀取數據前，存儲在內部緩存的最大字節(jié)數；默認=16kb， objectMode流是16.
  • encoding{String} 若指定，則Buffer會被解碼成所給編碼的字符串，默認為null。
  • objectMode{Boolean} 該流是否為對象的流。意思是說stream.read(n)返回一個單獨的值，而不是大小為n的Buffer。

Readable的擴展類中，確保調用了Readable的構造函數，這樣才能正確初始化。

readable._read(size)

• size{Number} 異步讀取的字節(jié)數

注意：實現這個函數，但不要直接調用。

這個函數不要直接調用。在子類里實現，僅能被內部的Readable類調用。

所有可讀流（Readable stream） 的實現必須停供一個_read方法，從底層資源里獲取數據。

這個方法以下劃線開頭，是因為對于定義它的類是內部的，不會被用戶程序直接調用。你可以在自己的擴展類中實現。

當數據可用時，通過調用readable.push(chunk)將之放到讀取隊列中。再次調用_read，需要繼續(xù)推出更多數據。

size參數僅供參考。調用“read”可以知道知道應當抓取多少數據；其余與之無關的實現，比如TCP或TLS，則可忽略這個參數，并在可用時返回數據。例如，沒有必要“等到”size個字節(jié)可用時才調用stream.push(chunk)。

readable.push(chunk[, encoding])

• chunk {Buffer | null | String} 推入到讀取隊列的數據塊
• encoding {String} 字符串塊的編碼。必須是有效的Buffer編碼，比如utf8或ascii。
• 返回{Boolean}是否應該繼續(xù)推入

注意: 這個函數必須被 Readable 實現者調用, 而不是可讀流（Readable stream）的消費者.

_read()函數直到調用push(chunk)后才能被再次調用。

Readable類將數據放到讀取隊列，當'readable'事件觸發(fā)后，被read()方法取出。push()方法會插入數據到讀取隊列中。如果調用了null，會觸發(fā)數據結束信號 (EOF)。

這個API被設計成盡可能地靈活。比如說，你可以包裝一個低級別的，具備某種暫停/恢復機制，和數據回調的數據源。這種情況下，你可以通過這種方式包裝低級別來源對象：

javascript
// source is an object with readStop() and readStart() 方法s,
// and an `ondata` member that gets called when it has data, and
// an `onend` member that gets called when the data is over.

util.inherits(SourceWrapper, Readable);

function SourceWrapper(options) {
  Readable.call(this, options);

  this._source = getLowlevelSourceObject();
  var self = this;

  // Every time there's data, we push it into the internal buffer.
  this._source.ondata = function(chunk) {
    // if push() 返回 false, then we need to stop reading from source
    if (!self.push(chunk))
      self._source.readStop();
  };

  // When the source ends, we push the EOF-signaling `null` chunk
  this._source.onend = function() {
    self.push(null);
  };
}

// _read will be called when the stream wants to pull more data in
// the advisory size 參數 is ignored in this case.
SourceWrapper.prototype._read = function(size) {
  this._source.readStart();
};

類： stream.Writable

stream.Writable是個抽象類，它擴展了一個底層的實現_write(chunk, encoding, callback)方法.

參考上面的流實現程序API，來了解在你的程序里如何消費可寫流。下面內容介紹了如何在你的程序里實現可寫流。

new stream.Writable([options])

• options {Object}
  • highWaterMark {Number} 當write()返回false時的緩存級別。默認=16kb，objectMode流是16。
  • decodeStrings {Boolean} 傳給_write()前是否解碼為字符串。默認=true
  • objectMode {Boolean}write(anyObj)是否是有效操作；如果為true，可以寫任意數據，而不僅僅是Buffer/String。默認=false

請確保Writable類的擴展類中，調用構造函數以便緩沖設定能被正確初始化。

writable._write(chunk, encoding, callback)

• chunk {Buffer | String} 要寫入的數據塊?？偸莃uffer， 除非decodeStrings選項為false。
• encoding {String} 如果數據塊是字符串，這個參數就是編碼方式。如果是緩存，則忽略。注意，除非decodeStrings被設置為false，否則這個數據塊一直是buffer。
• callback{函數} 當你處理完數據后調用這個函數 (錯誤參數為可選參數)。

所以可寫流（Writable stream ） 實現必須提供一個_write()方法，來發(fā)送數據給底層資源。

注意: 這個函數不能直接調用，由子類實現，僅內部可寫方法可以調用。

使用標準的callback(error)方法調用回調函數，來表明寫入完成或遇到錯誤。

如果構造函數選項中設定了decodeStrings標識，則chunk可能會是字符串而不是Buffer，encoding表明了字符串的格式。這種設計是為了支持對某些字符串數據編碼提供優(yōu)化處理的實現。如果你沒有明確的設置decodeStrings為false，這樣你就可以安不管encoding參數，并假定chunk一直是一個緩存。

該方法以下劃線開頭，是因為對于定義它的類來說，這個方法是內部的，并且不應該被用戶程序直接調用。你應當在你的擴充類中重寫這個方法。

writable._writev(chunks, callback)

• chunks {Array} 準備寫入的數據塊，每個塊格式如下:{ chunk: ..., encoding: ... }.
• callback {函數} 當你處理完數據后調用這個函數 (錯誤參數為可選參數)。

注意: 這個函數不能直接調用。由子類實現，僅內部可寫方法可以調用.

這個函數的實現是可選的。多數情況下，沒有必要實現。如果實現，將會在所有數據塊緩存到寫隊列后調用。

類： stream.Duplex

雙工流（duplex stream）同時兼具可讀和可寫特性，比如一個TCP socket連接。

注意stream.Duplex可以像Readable或Writable一樣被擴充，實現了底層_read(sise) 和_write(chunk, encoding, callback) 方法的抽象類。

由于JavaScript并沒有多重繼承能力，因此這個類繼承自Readable，寄生自Writable.從而讓用戶在雙工擴展類中同時實現低級別的_read(n)方法和低級別的_write(chunk, encoding, callback)方法。

new stream.Duplex(options)

• options {Object} 傳遞Writable and Readable構造函數，有以下的內容：
  • allowHalfOpen {Boolean} 默認=true。 如果設置為false，當寫端結束的時候，流會自動的結束讀端，反之亦然。
  • readableObjectMode {Boolean} 默認=false。將objectMode設為讀端的流，如果為true，將沒有效果。
  • writableObjectMode {Boolean} 默認=false。將objectMode設為寫端的流，如果為true，將沒有效果。

擴展自Duplex的類，確保調用了父親的構造函數，保證緩存設置能正確初始化。

類： stream.Transform

轉換流（transform class) 是雙工流（duplex stream），輸入輸出端有因果關系，比如，zlib流或crypto流。

輸入輸出沒有要求大小相同，塊數量相同，到達時間相同。例如，一個Hash流只會在輸入結束時產生一個數據塊的輸出；一個zlib流會產生比輸入小得多或大得多的輸出。

轉換流（transform class) 必須實現_transform()方法，而不是_read()和_write()方法，也可以實現_flush()方法（參見如下）。

new stream.Transform([options])

• options {Object} 傳遞給Writable和Readable構造函數。

擴展自轉換流（transform class) 的類，確保調用了父親的構造函數，保證緩存設置能正確初始化。

transform._transform(chunk, encoding, callback)

• chunk {Buffer | String} 準備轉換的數據塊。是buffer，除非decodeStrings選項設置為false。
• encoding {String} 如果數據塊是字符串, 這個參數就是編碼方式，否則就忽略這個參數
• callback {函數} 當你處理完數據后調用這個函數 (錯誤參數為可選參數)。

注意：這個函數不能直接調用。由子類實現，僅內部可寫方法可以調用.

所有的轉換流（transform class) 實現必須提供 _transform方法來接收輸入，并生產輸出。

_transform可以做轉換流（transform class)里的任何事，處理寫入的字節(jié)，傳給接口的寫端，異步I/O，處理事情等等。

調用transform.push(outputChunk)0次或多次，從這個輸入塊里產生輸出，依賴于你想要多少數據作為輸出。

僅在當前數據塊完全消費后調用這個回調。

注意，輸入塊可能有，也可能沒有對應的輸出塊。如果你提供了第二個參數，將會傳給push方法。如下述的例子：

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

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

該方法以下劃線開頭，是因為對于定義它的類來說，這個方法是內部的，并且不應該被用戶程序直接調用。你應當在你的擴充類中重寫這個方法。

transform._flush(callback)

• callback {函數} 當你處理完數據后調用這個函數 (錯誤參數為可選參數)

注意：這個函數不能直接調用。由子類實現，僅內部可寫方法可以調用.

某些情況下，轉換操作可能需要分發(fā)一點流最后的數據。例如，Zlib流會存儲一些內部狀態(tài)，以便優(yōu)化壓縮輸出。

有些時候，你可以實現_flush方法，它可以在最后面調用，當所有的寫入數據被消費后，分發(fā)end告訴讀端。和_transform一樣，當刷新操作完畢， transform.push(chunk)為0次或更多次數。

該方法以下劃線開頭，是因為對于定義它的類來說，這個方法是內部的，并且不應該被用戶程序直接調用。你應當在你的擴充類中重寫這個方法。

事件: 'finish' and 'end'

finish和end事件 分別來自Writable和Readable類。.end()事件結束后調用finish事件，所有的數據已經被_transform處理完畢，調用_flush后，所有的數據輸出完畢，觸發(fā)end。

Example:SimpleProtocolparser v2

上面的簡單協(xié)議分析例子列子可以通過使用高級別的Transform流來實現，和parseHeader ，SimpleProtocol v1列子類似。

在這個示例中，輸入會被導流到解析器中，而不是作為參數提供。這種做法更符合Node流的慣例。

javascript
var util = require('util');
var Transform = require('stream').Transform;
util.inherits(SimpleProtocol, Transform);

function SimpleProtocol(options) {
  if (!(this instanceof SimpleProtocol))
    return new SimpleProtocol(options);

  Transform.call(this, options);
  this._inBody = false;
  this._sawFirstCr = false;
  this._rawHeader = [];
  this.header = null;
}

SimpleProtocol.prototype._transform = function(chunk, encoding, done) {
  if (!this._inBody) {
    // check if the chunk has a \n\n
    var split = -1;
    for (var i = 0; i < chunk.length; i++) {
      if (chunk[i] === 10) { // '\n'
        if (this._sawFirstCr) {
          split = i;
          break;
        } else {
          this._sawFirstCr = true;
        }
      } else {
        this._sawFirstCr = false;
      }
    }

    if (split === -1) {
      // still waiting for the \n\n
      // stash the chunk, and try again.
      this._rawHeader.push(chunk);
    } else {
      this._inBody = true;
      var h = chunk.slice(0, split);
      this._rawHeader.push(h);
      var header = Buffer.concat(this._rawHeader).toString();
      try {
        this.header = JSON.parse(header);
      } catch (er) {
        this.emit('error', new Error('invalid simple protocol data'));
        return;
      }
      // and let them know that we are done parsing the header.
      this.emit('header', this.header);

      // now, because we got some extra data, emit this first.
      this.push(chunk.slice(split));
    }
  } else {
    // from there on, just provide the data to our consumer as-is.
    this.push(chunk);
  }
  done();
};

// Usage:
// var parser = new SimpleProtocol();
// source.pipe(parser)
// Now parser is 可讀流（Readable stream） that will emit 'header'
// with the parsed header data.

類： stream.PassThrough

這是Transform流的簡單實現，將輸入的字節(jié)簡單的傳遞給輸出。它的主要用途是測試和演示。偶爾要構建某種特殊流時也會用到。

流: 內部細節(jié)

緩沖

可寫流（Writable streams ）和可讀流（Readable stream）都會緩存數據到內部對象上，叫做_writableState.buffer或_readableState.buffer。

緩存的數據量，取決于構造函數是傳入的highWaterMark參數。

調用stream.push(chunk)時，緩存數據到可讀流（Readable stream）。在數據消費者調用stream.read()前，數據會一直緩存在內部隊列中。

調用stream.write(chunk)時，緩存數據到可寫流（Writable stream）。即使write()返回false。

流（尤其是pipe()方法）得目的是限制數據的緩存量到一個可接受的水平，使得不同速度的源和目的不會淹沒可用內存。

stream.read(0)

某些時候，你可能想不消費數據的情況下，觸發(fā)底層可讀流（Readable stream）機制的刷新。這種情況下可以調用stream.read(0)，它總會返回null。

如果內部讀取緩沖低于highWaterMark，并且流當前不在讀取狀態(tài)，那么調用read(0)會觸發(fā)一個低級_read調用。

雖然基本上沒有必要這么做。但你在Node內部的某些地方看到它確實這么做了，尤其是在Readable流類的內部。

stream.push('')

推一個0字節(jié)的字符串或緩存 (不在Object mode時)會發(fā)送有趣的副作用。 因為它是一個對stream.push()的調用，它將會結束reading進程。然而，它沒有添加任何數據到可讀緩沖區(qū)中，所以沒有東西可供用戶消費。

少數情況下，你當時沒有提供數據，但你的流的消費者（或你的代碼的其它部分）會通過調用stream.read(0)得知何時再次檢查。在這種情況下，你可以調用 stream.push('')。

到目前為止，這個功能唯一一個使用情景是在tls.CryptoStream類中，但它將在Node v0.12中被廢棄。如果你發(fā)現你不得不使用stream.push('')，請考慮另一種方式。

和老版本的兼容性

v0.10版本前，可讀流（Readable stream）接口比較簡單，因此功能和用處也小。

• 'data'事件會立即開始觸發(fā)，而不會等待你調用read()方法。如果你需要進行某些I/O來決定如何處理數據，那么你只能將數據塊儲存到某種緩沖區(qū)中以防它們流失。
• pause()方法僅供參考，而不保證生效。這意味著，即便流處于暫停狀態(tài)時，你仍然需要準備接收'data'事件。

在Node v0.10中, 加入了下文所述的Readable類。為了考慮向后兼容，添加了'data'事件監(jiān)聽器或resume()方法被調用時，可讀流（Readable stream）會切換到 "流動模式（flowing mode）"。其作用是，即便你不使用新的read()方法和'readable'事件，你也不必擔心丟失'data'數據塊。

大多數程序會維持正常功能。然而，下列條件下也會引入邊界情況：

• 沒有添加 ['data'事件][]處理器
• 從來沒有調用resume()方法
• 流從來沒有被倒流（pipe）到任何可寫目標上、

例如：

javascript
// WARNING!  BROKEN!
net.createServer(function(socket) {

  // we add an 'end' 方法, but never consume the data
  socket.on('end', function() {
    // It will never get here.
    socket.end('I got your message (but didnt read it)\n');
  });

}).listen(1337);

v0.10版本前的Node，流入的消息數據會被簡單的拋棄。之后的版本，socket會一直保持暫停。

這種情形下，調用resume()方法來開始工作：

javascript
// Workaround
net.createServer(function(socket) {

  socket.on('end', function() {
    socket.end('I got your message (but didnt read it)\n');
  });

  // start the flow of data, discarding it.
  socket.resume();

}).listen(1337);

可讀流（Readable stream）切換到流動模式（flowing mode），v0.10 版本前，可以使用wrap()方法將風格流包含在一個可讀類里。

Object Mode

通常情況下，流僅操作字符串和緩存。

處于object mode的流，除了緩存和字符串，還可以可以讀出普通JavaScript值。

在對象模式里，可讀流（Readable stream） 調用stream.read(size)總會返回單個項目，無論是什么參數。

在對象模式里， 可寫流（Writable stream ） 總會忽略傳給stream.write(data, encoding)的encoding參數。

特殊值null在對象模式里，依舊保持它的特殊性。也就說，對于對象模式的可讀流（Readable stream），stream.read()返回null意味著沒有更多數據，同時stream.push(null)會告知流數據結束（EOF）。

Node核心不存在對象模式的流，這種設計只被某些用戶態(tài)流式庫所使用。

應該在你的子類構造函數里，設置objectMode。在過程中設置不安全。

對于雙工流（Duplex streams），objectMode可以用readableObjectMode和writableObjectMode分別為讀寫端分別設置。這些選項，被轉換流（Transform streams）用來實現解析和序列化。

javascript
var util = require('util');
var StringDecoder = require('string_decoder').StringDecoder;
var Transform = require('stream').Transform;
util.inherits(JSONParseStream, Transform);

// Gets \n-delimited JSON  string data, and emits the parsed objects
function JSONParseStream() {
  if (!(this instanceof JSONParseStream))
    return new JSONParseStream();

  Transform.call(this, { readableObjectMode : true });

  this._buffer = '';
  this._decoder = new StringDecoder('utf8');
}

JSONParseStream.prototype._transform = function(chunk, encoding, cb) {
  this._buffer += this._decoder.write(chunk);
  // split on newlines
  var lines = this._buffer.split(/\r?\n/);
  // keep the last partial line buffered
  this._buffer = lines.pop();
  for (var l = 0; l < lines.length; l++) {
    var line = lines[l];
    try {
      var obj = JSON.parse(line);
    } catch (er) {
      this.emit('error', er);
      return;
    }
    // push the parsed object out to the readable consumer
    this.push(obj);
  }
  cb();
};

JSONParseStream.prototype._flush = function(cb) {
  // Just handle any leftover
  var rem = this._buffer.trim();
  if (rem) {
    try {
      var obj = JSON.parse(rem);
    } catch (er) {
      this.emit('error', er);
      return;
    }
    // push the parsed object out to the readable consumer
    this.push(obj);
  }
  cb();
};