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）接口是對你正在讀取的數(shù)據(jù)的來源的抽象。換句話說，數(shù)據(jù)來來自

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

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

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

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

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

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

如果沒有導(dǎo)流（pipe）目標(biāo)，調(diào)用 pause()方法.
如果有導(dǎo)流（pipe）目標(biāo), 移除所有的 ['data' 事件][]處理函數(shù), 調(diào)用 unpipe() 方法移除所有的導(dǎo)流（pipe）目標(biāo)。

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

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

事件: 'readable'

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

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

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

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

事件: 'data'

chunk {Buffer | String} 數(shù)據(jù)塊

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

如果你像盡快的從流中獲取數(shù)據(jù)，這是最快的方法。

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

事件: 'end'

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

注意，除非數(shù)據(jù)已經(jīng)被完全消費， the end 事件才會觸發(fā)。可以通過切換到流動模式（flowing mode）來實現(xiàn)，或者通過調(diào)用重復(fù)調(diào)用 read()獲取數(shù)據(jù)，直到結(jié)束。

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'

當(dāng)?shù)讓淤Y源（例如源頭的文件描述符）關(guān)閉時觸發(fā)。并不是所有流都會觸發(fā)這個事件。

事件: 'error'

{Error Object}

當(dāng)接收數(shù)據(jù)時發(fā)生錯誤觸發(fā)。

readable.read([size])

size {Number} 可選參數(shù)，需要讀入的數(shù)據(jù)量
返回 {String | Buffer | null}

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

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

如果你沒有指定 size 參數(shù)。將會返回內(nèi)部緩存的所有數(shù)據(jù)。

這個方法僅能再暫停模式（paused mode）里調(diào)用. 流動模式（flowing mode）下這個方法會被自動調(diào)用直到內(nèi)存緩存排空。

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

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

readable.setEncoding(encoding)

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

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

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

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）. 如果你不想從流中消費數(shù)據(jù)，而想得到end 事件，可以調(diào)用 readable.resume() 來打開數(shù)據(jù)流。

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ù)可用數(shù)據(jù)留在內(nèi)部緩沖區(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 是否被客戶端代碼明確的暫停（調(diào)用 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} 寫入數(shù)據(jù)的目標(biāo)
options {Object} 導(dǎo)流（pipe）選項
- end {Boolean} 讀取到結(jié)束符時，結(jié)束寫入者。默認(rèn) = true

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

可以導(dǎo)流到多個目標(biāo)。

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

這個函數(shù)返回目標(biāo)流, 因此你可以建立導(dǎo)流鏈：

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);

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

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

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

注意 process.stderr 和 process.stdout 直到進(jìn)程結(jié)束才會關(guān)閉，無論是否指定

readable.unpipe([destination])

destination {Writable Stream} 可選，指定解除導(dǎo)流的流

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

如果沒有指定 destination,所有的導(dǎo)流（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} 數(shù)據(jù)塊插入到讀隊列中

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

如果你在程序中必須經(jīng)常調(diào)用 stream.unshift(chunk) ，那你可以考慮實現(xiàn)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 流并未實現(xiàn)現(xiàn)在所有流的API（更多信息詳見下文“兼容性”章節(jié)）。

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

你應(yīng)該很少需要用到這個函數(shù)，但它會留下方便和舊版本的 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 ）接口是你正把數(shù)據(jù)寫到一個目標(biāo)的抽象。

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

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

chunk {String | Buffer} 準(zhǔn)備寫的數(shù)據(jù)
encoding {String} 編碼方式（如果chunk 是字符串）
callback {Function} 數(shù)據(jù)塊寫入后的回調(diào)
返回: {Boolean} 如果數(shù)據(jù)已被全部處理返回true

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

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

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

事件: 'drain'

如果調(diào)用 writable.write(chunk) 返回 false, drain 事件會告訴你什么時候?qū)⒏嗟臄?shù)據(jù)寫入到流中。

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()

強(qiáng)制緩存所有寫入。

調(diào)用 .uncork() 或 .end()后，會把緩存數(shù)據(jù)寫入。

writable.uncork()

寫入所有 .cork() 調(diào)用之后緩存的數(shù)據(jù)。

writable.setDefaultEncoding(encoding)

encoding {String} 新的默認(rèn)編碼
返回: Boolean

給寫數(shù)據(jù)流設(shè)置默認(rèn)編碼方式，如編碼有效，返回 true ，否則返回 false。

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

chunk {String | Buffer} 可選，要寫入的數(shù)據(jù)
encoding {String} 編碼方式（如果 chunk 是字符串）
callback {Function} 可選， stream 結(jié)束時的回調(diào)函數(shù)

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

調(diào)用 end() 后調(diào)用 write() 會產(chǎn)生錯誤。

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'

調(diào)用end() 方法后，并且所有的數(shù)據(jù)已經(jīng)寫入到底層系統(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} 是導(dǎo)流（pipe）到可寫流的源流

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

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} The source stream that unpiped this writable

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

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}

寫或?qū)Я鳎╬ipe）數(shù)據(jù)時，如果有錯誤會觸發(fā)。

類： stream.Duplex

雙工流（Duplex streams）是同時實現(xiàn)了 Readable and Writable 接口。用法詳見下文。

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

類： stream.Transform

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

轉(zhuǎn)換流（Transform streams）的例子包括:

API for Stream Implementors

無論實現(xiàn)什么形式的流，模式都是一樣的:

在你的子類中擴(kuò)展適合的父類. (util.inherits 方法很有幫助)
在你的構(gòu)造函數(shù)中調(diào)用父類的構(gòu)造函數(shù)，以確保內(nèi)部的機(jī)制初始化正確。
實現(xiàn)一個或多個方法，如下所列

所擴(kuò)展的類和要實現(xiàn)的方法取決于你要編寫的流類。

Use-case	Class	方法(s) to implement
Reading only	[Readable](#stream_class_stream_readable_1)	`[_read][]`
Writing only	[Writable](#stream_class_stream_writable_1)	`[_write][]`
Reading and writing	[Duplex](#stream_class_stream_duplex_1)	`[_read][]`, `[_write][]`
Operate on written data, then read the result	[Transform](#stream_class_stream_transform_1)	`_transform`, `_flush`

在你的代碼里，千萬不要調(diào)用 API for Stream Consumers 里的方法。否則可能會引起消費流的程序副作用。

類： stream.Readable

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

參照之前的API for Stream Consumers查看如何在你的程序里消費流。底下內(nèi)容解釋了在你的程序里如何實現(xiàn)可讀流（Readable stream）。

Example: 計數(shù)流

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

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 函數(shù)類似, 但它被實現(xiàn)為自定義流。注意這個實現(xiàn)不會將輸入數(shù)據(jù)轉(zhuǎn)換為字符串。

實際上，更好的辦法是將他實現(xiàn)為 Transform 流。下面的實現(xiàn)方法更好。

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} 停止從底層資源讀取數(shù)據(jù)前，存儲在內(nèi)部緩存的最大字節(jié)數(shù)。默認(rèn)=16kb, objectMode 流是16.
- encoding {String} 若指定，則 Buffer 會被解碼成所給編碼的字符串。缺省為 null
- objectMode {Boolean} 該流是否為對象的流。意思是說 stream.read(n) 返回一個單獨的值，而不是大小為 n 的 Buffer。

Readable 的擴(kuò)展類中，確保調(diào)用了 Readable 的構(gòu)造函數(shù)，這樣才能正確初始化。

readable._read(size)

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

注意: 實現(xiàn)這個函數(shù), 但不要直接調(diào)用.

這個函數(shù)不要直接調(diào)用. 在子類里實現(xiàn),僅能被內(nèi)部的 Readable 類調(diào)用。

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

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

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

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

readable.push(chunk[, encoding])

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

注意: 這個函數(shù)必須被 Readable 實現(xiàn)者調(diào)用, 而不是可讀流（Readable stream）的消費者.

_read() 函數(shù)直到調(diào)用push(chunk) 后才能被再次調(diào)用。

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

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

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 參數(shù) is ignored in this case.
SourceWrapper.prototype._read = function(size) {
  this._source.readStart();
};

類： stream.Writable

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

參考上面的API for Stream Consumers，來了解在你的程序里如何消費可寫流。下面內(nèi)容介紹了如何在你的程序里實現(xiàn)可寫流。

new stream.Writable([options])

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

請確保 Writable 類的擴(kuò)展類中，調(diào)用構(gòu)造函數(shù)以便緩沖設(shè)定能被正確初始化。

writable._write(chunk, encoding, callback)

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

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

注意: 這個函數(shù)不能直接調(diào)用 ,由子類實現(xiàn)，僅內(nèi)部可寫方法可以調(diào)用。

使用標(biāo)準(zhǔn)的 callback(error) 方法調(diào)用回調(diào)函數(shù)，來表明寫入完成或遇到錯誤。

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

該方法以下劃線開頭，是因為對于定義它的類來說，這個方法是內(nèi)部的，并且不應(yīng)該被用戶程序直接調(diào)用。你應(yīng)當(dāng)在你的擴(kuò)充類中重寫這個方法。

writable._writev(chunks, callback)

chunks {Array} 準(zhǔn)備寫入的數(shù)據(jù)塊，每個塊格式如下: { chunk: ..., encoding: ... }.
callback {函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個函數(shù) (錯誤參數(shù)為可選參數(shù))。

注意: 這個函數(shù)不能直接調(diào)用。 由子類實現(xiàn)，僅內(nèi)部可寫方法可以調(diào)用.

這個函數(shù)的實現(xiàn)是可選的。多數(shù)情況下，沒有必要實現(xiàn)。如果實現(xiàn)，將會在所有數(shù)據(jù)塊緩存到寫隊列后調(diào)用。

類： stream.Duplex

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

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

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

new stream.Duplex(options)

options {Object} 傳遞 Writable and Readable 構(gòu)造函數(shù)，有以下的內(nèi)容：
- allowHalfOpen {Boolean} 默認(rèn)=true. 如果設(shè)置為 false, 當(dāng)寫端結(jié)束的時候，流會自動的結(jié)束讀端，反之亦然。
- readableObjectMode {Boolean} 默認(rèn)=false. 將 objectMode 設(shè)為讀端的流，如果為 true，將沒有效果。
- writableObjectMode {Boolean} 默認(rèn)=false. 將 objectMode設(shè)為寫端的流，如果為 true，將沒有效果。

擴(kuò)展自 Duplex 的類，確保調(diào)用了父親的構(gòu)造函數(shù)，保證緩存設(shè)置能正確初始化。

類： stream.Transform

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

輸入輸出沒有要求大小相同，塊數(shù)量相同，到達(dá)時間相同。例如，一個 Hash 流只會在輸入結(jié)束時產(chǎn)生一個數(shù)據(jù)塊的輸出；一個 zlib 流會產(chǎn)生比輸入小得多或大得多的輸出。

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

new stream.Transform([options])

options {Object} 傳遞給 Writable and Readable 構(gòu)造函數(shù)。

擴(kuò)展自轉(zhuǎn)換流（transform class) 的類，確保調(diào)用了父親的構(gòu)造函數(shù)，保證緩存設(shè)置能正確初始化。

transform._transform(chunk, encoding, callback)

chunk {Buffer | String} 準(zhǔn)備轉(zhuǎn)換的數(shù)據(jù)塊。是buffer，除非 decodeStrings 選項設(shè)置為 false。
encoding {String} 如果數(shù)據(jù)塊是字符串, 這個參數(shù)就是編碼方式，否則就忽略這個參數(shù)
callback {函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個函數(shù) (錯誤參數(shù)為可選參數(shù))。

注意: 這個函數(shù)不能直接調(diào)用。 由子類實現(xiàn)，僅內(nèi)部可寫方法可以調(diào)用.

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

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

調(diào)用 transform.push(outputChunk)0或多次，從這個輸入塊里產(chǎn)生輸出，依賴于你想要多少數(shù)據(jù)作為輸出。

僅在當(dāng)前數(shù)據(jù)塊完全消費后調(diào)用這個回調(diào)。注意，輸入塊可能有，也可能沒有對應(yīng)的輸出塊。如果你提供了第二個參數(shù)，將會傳給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 {函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個函數(shù) (錯誤參數(shù)為可選參數(shù))

注意: 這個函數(shù)不能直接調(diào)用。 由子類實現(xiàn)，僅內(nèi)部可寫方法可以調(diào)用.

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

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

事件: 'finish' and 'end'

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

Example: `SimpleProtocol` parser v2

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

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

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

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


            上一篇：定時器下一篇：域

流

面向流消費者的 API

類： stream.Readable

事件: 'readable'

事件: 'data'

事件: 'end'

事件: 'close'

事件: 'error'

readable.read([size])

readable.setEncoding(encoding)

readable.resume()

readable.pause()

readable.isPaused()

readable.pipe(destination[, options])

readable.unpipe([destination])

readable.unshift(chunk)

readable.wrap(stream)

類： stream.Writable

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

事件: 'drain'

writable.cork()

writable.uncork()

writable.setDefaultEncoding(encoding)

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

事件: 'finish'

事件: 'pipe'

事件: 'unpipe'

事件: 'error'

類： stream.Duplex

類： stream.Transform

API for Stream Implementors

類： stream.Readable

Example: 計數(shù)流

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

new stream.Readable([options])

readable._read(size)

readable.push(chunk[, encoding])

類： stream.Writable

new stream.Writable([options])

writable._write(chunk, encoding, callback)

writable._writev(chunks, callback)

類： stream.Duplex

new stream.Duplex(options)

類： stream.Transform

new stream.Transform([options])

transform._transform(chunk, encoding, callback)

transform._flush(callback)

事件: 'finish' and 'end'

Example: SimpleProtocol parser v2

Example: `SimpleProtocol` parser v2