node.js之stream模塊

stream

流是一個(gè)抽象接口,在 Node 里被不同的對(duì)象實(shí)現(xiàn)娱俺。例如 request to an HTTP server 是流缎谷,stdout 是流井濒。流是可讀,可寫,或者可讀寫瑞你。所有的流是 EventEmitter 的實(shí)例酪惭。

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

這個(gè)文檔分為 3 個(gè)章節(jié)虏缸。第一個(gè)章節(jié)解釋了在你的程序中使用流時(shí)候需要了解的部分甥厦。如果你不用實(shí)現(xiàn)流式 API,可以只看這個(gè)章節(jié)寇钉。

如果你想實(shí)現(xiàn)你自己的流刀疙,第二個(gè)章節(jié)解釋了這部分 API。這些 API 讓你的實(shí)現(xiàn)更加簡(jiǎn)單扫倡。

第三個(gè)部分深入的解釋了流是如何工作的谦秧,包括一些內(nèi)部機(jī)制和函數(shù),這些內(nèi)容不要改動(dòng)撵溃,除非你明確知道你要做什么疚鲤。

面向流消費(fèi)者的 API
流可以是可讀(Readable),可寫(Writable)缘挑,或者兼具兩者(Duplex集歇,雙工)的。

所有的流都是事件分發(fā)器(EventEmitters)语淘,但是也有自己的方法和屬性诲宇,這取決于他它們是可讀(Readable),可寫(Writable)惶翻,或者兼具兩者(Duplex姑蓝,雙工)的。

如果流式可讀寫的吕粗,則它實(shí)現(xiàn)了下面的所有方法和事件浇衬。因此祖秒,這個(gè)章節(jié) API 完全闡述了Duplex 或 Transform 流寥殖,即便他們的實(shí)現(xiàn)有所不同推沸。

沒有必要為了消費(fèi)流而在你的程序里實(shí)現(xiàn)流的接口。如果你正在你的程序里實(shí)現(xiàn)流接口议泵,請(qǐng)同時(shí)參考下面的API for Stream Implementors占贫。

基本所有的 Node 程序,無論多簡(jiǎn)單肢簿,都會(huì)使用到流靶剑。這有一個(gè)使用流的例子蜻拨。

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)接口是對(duì)你正在讀取的數(shù)據(jù)的來源的抽象池充。換句話說桩引,數(shù)據(jù)來來自

可讀流(Readable stream)不會(huì)分發(fā)數(shù)據(jù),直到你表明準(zhǔn)備就緒收夸。

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

注意: 如果沒有綁定數(shù)據(jù)處理函數(shù)厘灼,并且沒有 pipe() 目標(biāo),流會(huì)切換到流動(dòng)模式(flowing mode)咽瓷,并且數(shù)據(jù)會(huì)丟失设凹。

可以通過下面幾個(gè)方法,將流切換到流動(dòng)模式(flowing mode)茅姜。

添加一個(gè) '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)聽器并不會(huì)自動(dòng)暫停流。同樣的素标,當(dāng)有導(dǎo)流目標(biāo)時(shí)称诗,調(diào)用 pause() 并不能保證流在那些目標(biāo)排空后,請(qǐng)求更多數(shù)據(jù)時(shí)保持暫停狀態(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'

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

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

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

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

事件: 'data'

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

綁定一個(gè) data 事件的監(jiān)聽器(listener)到一個(gè)未明確暫停的流享潜,會(huì)將流切換到流動(dòng)模式困鸥。數(shù)據(jù)會(huì)盡額能的傳遞。

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

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

事件: 'end'

如果沒有更多的可讀數(shù)據(jù),將會(huì)觸發(fā)這個(gè)事件艺蝴。

注意猬腰,除非數(shù)據(jù)已經(jīng)被完全消費(fèi), the end 事件才會(huì)觸發(fā)猜敢。 可以通過切換到流動(dòng)模式(flowing mode)來實(shí)現(xiàn)姑荷,或者通過調(diào)用重復(fù)調(diào)用 read()獲取數(shù)據(jù)盒延,直到結(jié)束。

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)閉時(shí)觸發(fā)鼠冕。并不是所有流都會(huì)觸發(fā)這個(gè)事件添寺。

事件: 'error'

{Error Object}當(dāng)接收數(shù)據(jù)時(shí)發(fā)生錯(cuò)誤觸發(fā)。

readable.read([size])

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

read() 方法從內(nèi)部緩存中拉取數(shù)據(jù)计露。如果沒有可用數(shù)據(jù),將會(huì)返回null
如果傳了 size參數(shù)憎乙,將會(huì)返回相當(dāng)字節(jié)的數(shù)據(jù)票罐。如果size不可用,將會(huì)返回 null
如果你沒有指定 size 參數(shù)泞边。將會(huì)返回內(nèi)部緩存的所有數(shù)據(jù)该押。
這個(gè)方法僅能再暫停模式(paused mode)里調(diào)用. 流動(dòng)模式(flowing mode)下這個(gè)方法會(huì)被自動(dòng)調(diào)用直到內(nèi)存緩存排空。

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

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

readable.setEncoding(encoding)

encoding {String} 要使用的編碼.
返回: this
調(diào)用此函數(shù)會(huì)使得流返回指定編碼的字符串阵谚,而不是 Buffer 對(duì)象蚕礼。例如,如果你調(diào)用readable.setEncoding('utf8')椭蹄,輸出數(shù)據(jù)將會(huì)是UTF-8 編碼闻牡,并且返回字符串。如果你調(diào)用 readable.setEncoding('hex')绳矩,將會(huì)返回2進(jìn)制編碼的數(shù)據(jù)罩润。

該方法能正確處理多字節(jié)字符。如果不想這么做翼馆,僅簡(jiǎn)單的直接拉取緩存并調(diào)buf.toString(encoding) 割以,可能會(huì)導(dǎo)致字節(jié)錯(cuò)位。因此应媚,如果你想以字符串讀取數(shù)據(jù)严沥,請(qǐng)使用這個(gè)方法。

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
這個(gè)方法讓可讀流(Readable stream)繼續(xù)觸發(fā) data 事件.

這個(gè)方法會(huì)將流切換到流動(dòng)模式(flowing mode). 如果你不想從流中消費(fèi)數(shù)據(jù)中姜,而想得到end 事件消玄,可以調(diào)用 [readable.resume()][] 來打開數(shù)據(jù)流。

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

返回: this
這個(gè)方法會(huì)使得流動(dòng)模式(flowing mode)的流停止觸發(fā) data 事件, 切換到流動(dòng)模式(flowing mode). 并讓后續(xù)可用數(shù)據(jù)留在內(nèi)部緩沖區(qū)中丢胚。

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
這個(gè)方法返回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) 選項(xiàng)
  • end {Boolean} 讀取到結(jié)束符時(shí),結(jié)束寫入者携龟。默認(rèn) = true

這個(gè)方法從可讀流(Readable stream)拉取所有數(shù)據(jù), 并將數(shù)據(jù)寫入到提供的目標(biāo)中兔跌。自動(dòng)管理流量,這樣目標(biāo)不會(huì)快速的可讀流(Readable stream)淹沒峡蟋。

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

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

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

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

例如, 模擬 Unix 的 cat 命令:

process.stdin.pipe(process.stdout);

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

這會(huì)讓 writer保持打開狀態(tài)方灾,可以在最后寫入"Goodbye" 建蹄。

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

注意 process.stderrprocess.stdout 直到進(jìn)程結(jié)束才會(huì)關(guān)閉碌更,無論是否指定

readable.unpipe([destination])

  • destination {Writable Stream} 可選裕偿,指定解除導(dǎo)流的流

這個(gè)方法會(huì)解除之前調(diào)用 pipe() 設(shè)置的鉤子( pipe() )。
如果沒有指定 destination,所有的 導(dǎo)流(pipe) 都會(huì)被移除痛单。
如果指定了 destination嘿棘,但是沒有建立如果沒有指定 destination,則什么事情都不會(huì)發(fā)生旭绒。

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ù)塊插入到讀隊(duì)列中

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

如果你在程序中必須經(jīng)常調(diào)用 stream.unshift(chunk) 忽匈,那你可以考慮實(shí)現(xiàn) Transform 來替換(參見下文API for Stream Implementors)房午。

// 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} 一個(gè)舊式的可讀流(Readable stream)

v0.10 版本之前的 Node 流并未實(shí)現(xiàn)現(xiàn)在所有流的API(更多信息詳見下文“兼容性”章節(jié))。

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

你應(yīng)該很少需要用到這個(gè)函數(shù),但它會(huì)留下方便和舊版本的 Node 程序和庫交互批狐。

例如:

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ù)寫到一個(gè)目標(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

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

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

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

事件: 'drain'

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

// 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()后芝此,會(huì)把緩存數(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é)束時(shí)的回調(diào)函數(shù)

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

調(diào)用 end() 后調(diào)用 write() 會(huì)產(chǎn)生錯(cuò)誤非区。

// 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)盹廷,將會(huì)觸發(fā)這個(gè)事件征绸。

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)到可寫流的源流

無論何時(shí)在可寫流(Writable stream )上調(diào)用pipe() 方法,都會(huì)觸發(fā) 'pipe' 事件速和,添加這個(gè)流到目標(biāo)歹垫。

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

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

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ù)時(shí)颠放,如果有錯(cuò)誤會(huì)觸發(fā)排惨。

類: stream.Duplex

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

雙工流(Duplex streams) 的例子包括:

  • tcp sockets
  • zlib streams
  • crypto streams

類: stream.Transform

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

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

  • zlib streams
  • crypto streams

API for Stream Implementors

無論實(shí)現(xiàn)什么形式的流欲低,模式都是一樣的:

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

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

<table>
<thead>
<tr>
<th>
<p>Use-case</p>
</th>
<th>
<p>Class</p>
</th>
<th>
<p>方法(s) to implement</p>
</th>
</tr>
</thead>
<tr>
<td>
<p>Reading only</p>
</td>
<td>
<p>Readable</p>
</td>
<td>
<p><code>_read</code></p>
</td>
</tr>
<tr>
<td>
<p>Writing only</p>
</td>
<td>
<p>Writable</p>
</td>
<td>
<p><code>_write</code></p>
</td>
</tr>
<tr>
<td>
<p>Reading and writing</p>
</td>
<td>
<p>Duplex</p>
</td>
<td>
<p><code>_read</code>, <code>_write</code></p>
</td>
</tr>
<tr>
<td>
<p>Operate on written data, then read the result</p>
</td>
<td>
<p>Transform</p>
</td>
<td>
<p><code>_transform</code>, <code>_flush</code></p>
</td>
</tr>
</table>

在你的代碼里,千萬不要調(diào)用 API for Stream Consumers 里的方法腊瑟。否則可能會(huì)引起消費(fèi)流的程序副作用聚假。

類: stream.Readable

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

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

Example: 計(jì)數(shù)流

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

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: 簡(jiǎn)單協(xié)議 v1 (初始版)

和之前描述的 parseHeader 函數(shù)類似, 但它被實(shí)現(xiàn)為自定義流纱控。注意這個(gè)實(shí)現(xiàn)不會(huì)將輸入數(shù)據(jù)轉(zhuǎn)換為字符串。

實(shí)際上菜秦,更好的辦法是將他實(shí)現(xiàn)為 Transform 流甜害。下面的實(shí)現(xiàn)方法更好。

// 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 讀取隊(duì)列 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ù)前喷户,存儲(chǔ)在內(nèi)部緩存的最大字節(jié)數(shù)唾那。默認(rèn)=16kb, objectMode 流是16.
  • encoding {String} 若指定访锻,則 Buffer 會(huì)被解碼成所給編碼的字符串褪尝。缺省為 null
  • objectMode {Boolean} 該流是否為對(duì)象的流。意思是說 stream.read(n) 返回一個(gè)單獨(dú)的值期犬,而不是大小為 n 的 Buffer河哑。

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

readable._read(size)

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

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

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

所有可讀流(Readable stream) 的實(shí)現(xiàn)必須停供一個(gè) _read 方法鲤妥,從底層資源里獲取數(shù)據(jù)佳吞。

這個(gè)方法以下劃線開頭,是因?yàn)閷?duì)于定義它的類是內(nèi)部的棉安,不會(huì)被用戶程序直接調(diào)用底扳。 你可以在自己的擴(kuò)展類中實(shí)現(xiàn)。

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

size 參數(shù)僅供參考. 調(diào)用 “read” 可以知道知道應(yīng)當(dāng)抓取多少數(shù)據(jù)阱冶;其余與之無關(guān)的實(shí)現(xiàn),比如 TCP 或 TLS滥嘴,則可忽略這個(gè)參數(shù)木蹬,并在可用時(shí)返回?cái)?shù)據(jù)。例如若皱,沒有必要“等到” size 個(gè)字節(jié)可用時(shí)才調(diào)用 stream.push(chunk)镊叁。

readable.push(chunk[, encoding])

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

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

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

Readable 類將數(shù)據(jù)放到讀取隊(duì)列,當(dāng) 'readable' 事件觸發(fā)后蛔添,被 read() 方法取出痰催。push() 方法會(huì)插入數(shù)據(jù)到讀取隊(duì)列中。如果調(diào)用了 null 迎瞧,會(huì)觸發(fā) 數(shù)據(jù)結(jié)束信號(hào) (EOF)夸溶。

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

// 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 是個(gè)抽象類,它擴(kuò)展了一個(gè)底層的實(shí)現(xiàn) _write(chunk, encoding, callback) 方法.

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

new stream.Writable([options])

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

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

writable._write(chunk, encoding, callback)

  • chunk {Buffer | String} 要寫入的數(shù)據(jù)塊〖鸷辏總是 buffer沈贝, 除非 decodeStrings 選項(xiàng)為 false
  • encoding {String} 如果數(shù)據(jù)塊是字符串蚀浆,這個(gè)參數(shù)就是編碼方式缀程。如果是緩存,則忽略市俊。注意杨凑,除非decodeStrings 被設(shè)置為 false ,否則這個(gè)數(shù)據(jù)塊一直是buffer摆昧。
  • callback {函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個(gè)函數(shù) (錯(cuò)誤參數(shù)為可選參數(shù))撩满。

所以可寫流(Writable stream ) 實(shí)現(xiàn)必須提供一個(gè) _write()方法,來發(fā)送數(shù)據(jù)給底層資源。
注意: 這個(gè)函數(shù)不能直接調(diào)用 ,由子類實(shí)現(xiàn)伺帘, 僅內(nèi)部可寫方法可以調(diào)用昭躺。
使用標(biāo)準(zhǔn)的 callback(error) 方法調(diào)用回調(diào)函數(shù),來表明寫入完成或遇到錯(cuò)誤伪嫁。

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

該方法以下劃線開頭,是因?yàn)閷?duì)于定義它的類來說龙助,這個(gè)方法是內(nèi)部的砰奕,并且不應(yīng)該被用戶程序直接調(diào)用。你應(yīng)當(dāng)在你的擴(kuò)充類中重寫這個(gè)方法提鸟。

writable._writev(chunks, callback)

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

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

這個(gè)函數(shù)的實(shí)現(xiàn)是可選的。多數(shù)情況下铣缠,沒有必要實(shí)現(xiàn)。如果實(shí)現(xiàn)昆禽,將會(huì)在所有數(shù)據(jù)塊緩存到寫隊(duì)列后調(diào)用蝗蛙。

類:stream.Duplex


雙工流(duplex stream)同時(shí)兼具可讀和可寫特性,比如一個(gè) TCP socket 連接醉鳖。
注意 stream.Duplex 可以像 Readable 或 Writable 一樣被擴(kuò)充捡硅,實(shí)現(xiàn)了底層 _read(sise)_write(chunk, encoding, callback) 方法的抽象類。

由于 JavaScript 并沒有多重繼承能力盗棵,因此這個(gè)類繼承自 Readable壮韭,寄生自 Writable.從而讓用戶在雙工擴(kuò)展類中同時(shí)實(shí)現(xiàn)低級(jí)別的_read(n) 方法和低級(jí)別的 _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é)束的時(shí)候喷屋,流會(huì)自動(dòng)的結(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á)時(shí)間相同。例如笙以,一個(gè) Hash 流只會(huì)在輸入結(jié)束時(shí)產(chǎn)生一個(gè)數(shù)據(jù)塊的輸出淌实;一個(gè) zlib 流會(huì)產(chǎn)生比輸入小得多或大得多的輸出。

轉(zhuǎn)換流(transform class) 必須實(shí)現(xiàn)_transform() 方法猖腕,而不是_read()_write() 方法拆祈,也可以實(shí)現(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 選項(xiàng)設(shè)置為 false蜡豹。
  • encoding {String} 如果數(shù)據(jù)塊是字符串, 這個(gè)參數(shù)就是編碼方式麸粮,否則就忽略這個(gè)參數(shù)
  • callback {函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個(gè)函數(shù) (錯(cuò)誤參數(shù)為可選參數(shù))。

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

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

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

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

僅在當(dāng)前數(shù)據(jù)塊完全消費(fèi)后調(diào)用這個(gè)回調(diào)彰檬。注意,輸入塊可能有谎砾,也可能沒有對(duì)應(yīng)的輸出塊逢倍。如果你提供了第二個(gè)參數(shù),將會(huì)傳給push 方法景图。如底下的例子

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

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

該方法以下劃線開頭较雕,是因?yàn)閷?duì)于定義它的類來說,這個(gè)方法是內(nèi)部的挚币,并且不應(yīng)該被用戶程序直接調(diào)用亮蒋。你應(yīng)當(dāng)在你的擴(kuò)充類中重寫這個(gè)方法。

transform._flush(callback)

  • callback {函數(shù)} 當(dāng)你處理完數(shù)據(jù)后調(diào)用這個(gè)函數(shù) (錯(cuò)誤參數(shù)為可選參數(shù))

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

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

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

該方法以下劃線開頭拱她,是因?yàn)閷?duì)于定義它的類來說,這個(gè)方法是內(nèi)部的扔罪,并且不應(yīng)該被用戶程序直接調(diào)用。你應(yīng)當(dāng)在你的擴(kuò)充類中重寫這個(gè)方法桶雀。

事件: 'finish' and 'end'

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

Example: SimpleProtocol parser v2

上面的簡(jiǎn)單協(xié)議分析例子列子可以通過使用高級(jí)別的[Transform][] 流來實(shí)現(xiàn)辜腺,和 parseHeaderSimpleProtocol v1列子類似。

在這個(gè)示例中评疗,輸入會(huì)被導(dǎo)流到解析器中测砂,而不是作為參數(shù)提供。這種做法更符合 Node 流的慣例百匆。

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ǎn)單實(shí)現(xiàn)砌些,將輸入的字節(jié)簡(jiǎn)單的傳遞給輸出。它的主要用途是測(cè)試和演示加匈。偶爾要構(gòu)建某種特殊流時(shí)也會(huì)用到存璃。

流: 內(nèi)部細(xì)節(jié)

緩沖

可寫流(Writable streams ) 和 可讀流(Readable stream)都會(huì)緩存數(shù)據(jù)到內(nèi)部對(duì)象上,叫做 _writableState.buffer_readableState.buffer雕拼。

緩存的數(shù)據(jù)量纵东,取決于構(gòu)造函數(shù)是傳入的 highWaterMark 參數(shù)。

調(diào)用 stream.push(chunk) 時(shí)啥寇,緩存數(shù)據(jù)到可讀流(Readable stream)偎球。在數(shù)據(jù)消費(fèi)者調(diào)用 stream.read() 前,數(shù)據(jù)會(huì)一直緩存在內(nèi)部隊(duì)列中示姿。

調(diào)用 stream.write(chunk) 時(shí)甜橱,緩存數(shù)據(jù)到可寫流(Writable stream)。即使 write() 返回 false栈戳。

流(尤其是pipe() 方法)得目的是限制數(shù)據(jù)的緩存量到一個(gè)可接受的水平岂傲,使得不同速度的源和目的不會(huì)淹沒可用內(nèi)存。

stream.read(0)

某些時(shí)候子檀,你可能想不消費(fèi)數(shù)據(jù)的情況下镊掖,觸發(fā)底層可讀流(Readable stream)機(jī)制的刷新。這種情況下可以調(diào)用 stream.read(0)褂痰,它總會(huì)返回 null亩进。

如果內(nèi)部讀取緩沖低于 highWaterMark,并且流當(dāng)前不在讀取狀態(tài)缩歪,那么調(diào)用 read(0) 會(huì)觸發(fā)一個(gè)低級(jí) _read 調(diào)用归薛。

雖然基本上沒有必要這么做。但你在 Node 內(nèi)部的某些地方看到它確實(shí)這么做了匪蝙,尤其是在 Readable 流類的內(nèi)部主籍。

stream.push('')

推一個(gè)0字節(jié)的字符串或緩存 (不在Object mode時(shí))會(huì)發(fā)送有趣的副作用. 因?yàn)樗且粋€(gè)對(duì)
stream.push() 的調(diào)用, 它將會(huì)結(jié)束 reading 進(jìn)程. 然而,它沒有添加任何數(shù)據(jù)到可讀緩沖區(qū)中逛球,所以沒有東西可供用戶消費(fèi)千元。

少數(shù)情況下,你當(dāng)時(shí)沒有提供數(shù)據(jù)颤绕,但你的流的消費(fèi)者(或你的代碼的其它部分)會(huì)通過調(diào)用 stream.read(0) 得知何時(shí)再次檢查幸海。在這種情況下祟身,你可以調(diào)用 stream.push('')

到目前為止物独,這個(gè)功能唯一一個(gè)使用情景是在 tls.CryptoStream 類中袜硫,但它將在 Node v0.12 中被廢棄。如果你發(fā)現(xiàn)你不得不使用 stream.push('')议纯,請(qǐng)考慮另一種方式父款。

和老版本的兼容性

v0.10 版本前,可讀流(Readable stream)接口比較簡(jiǎn)單瞻凤,因此功能和用處也小憨攒。

  • 'data'事件會(huì)立即開始觸發(fā),而不會(huì)等待你調(diào)用 read() 方法阀参。如果你需要進(jìn)行某些 I/O 來決定如何處理數(shù)據(jù)肝集,那么你只能將數(shù)據(jù)塊儲(chǔ)存到某種緩沖區(qū)中以防它們流失。
  • pause() 方法僅供參考蛛壳,而不保證生效杏瞻。這意味著,即便流處于暫停狀態(tài)時(shí)衙荐,你仍然需要準(zhǔn)備接收 'data' 事件捞挥。

在 Node v0.10中, 加入了下文所述的 Readable 類。為了考慮向后兼容忧吟,添加了 'data' 事件監(jiān)聽器或 resume() 方法被調(diào)用時(shí)砌函,可讀流(Readable stream)會(huì)切換到 "流動(dòng)模式(flowing mode)"。其作用是溜族,即便你不使用新的 read() 方法和'readable'事件讹俊,你也不必?fù)?dān)心丟失'data' 數(shù)據(jù)塊。

大多數(shù)程序會(huì)維持正常功能煌抒。然而仍劈,下列條件下也會(huì)引入邊界情況:

  • 沒有添加 'data' 事件 處理器
  • 從來沒有調(diào)用 resume() 方法
  • 流從來沒有被倒流(pipe)到任何可寫目標(biāo)上、

例如:

// 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, 流入的消息數(shù)據(jù)會(huì)被簡(jiǎn)單的拋棄寡壮。之后的版本贩疙,socket 會(huì)一直保持暫停。

這種情形下况既,調(diào)用resume() 方法來開始工作:

// 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)切換到流動(dòng)模式(flowing mode),v0.10 版本前屋群,可以使用wrap() 方法將風(fēng)格流包含在一個(gè)可讀類里。

Object Mode

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

處于 object mode 的流邪乍,除了 緩存和字符串降狠,還可以可以讀出普通 JavaScript值对竣。

在對(duì)象模式里,可讀流(Readable stream) 調(diào)用 stream.read(size)總會(huì)返回單個(gè)項(xiàng)目榜配,無論是什么參數(shù)否纬。

在對(duì)象模式里, 可寫流(Writable stream ) 總會(huì)忽略傳給stream.write(data, encoding)encoding參數(shù)蛋褥。

特殊值 null 在對(duì)象模式里临燃,依舊保持它的特殊性。也就說烙心,對(duì)于對(duì)象模式的可讀流(Readable stream)膜廊,stream.read() 返回 null 意味著沒有更多數(shù)據(jù),同時(shí)stream.push(null) 會(huì)告知流數(shù)據(jù)結(jié)束(EOF)淫茵。

Node 核心不存在對(duì)象模式的流爪瓜,這種設(shè)計(jì)只被某些用戶態(tài)流式庫所使用。

應(yīng)該在你的子類構(gòu)造函數(shù)里匙瘪,設(shè)置objectMode 铆铆。在過程中設(shè)置不安全。

對(duì)于雙工流(Duplex streams)丹喻,objectMode可以用readableObjectModewritableObjectMode 分別為讀寫端分別設(shè)置薄货。這些選項(xiàng)攀隔,被轉(zhuǎn)換流(Transform streams)用來實(shí)現(xiàn)解析和序列化蝇狼。

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(); 
  };
最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
  • 序言:七十年代末,一起剝皮案震驚了整個(gè)濱河市错英,隨后出現(xiàn)的幾起案子骑冗,更是在濱河造成了極大的恐慌赊瞬,老刑警劉巖,帶你破解...
    沈念sama閱讀 207,113評(píng)論 6 481
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件贼涩,死亡現(xiàn)場(chǎng)離奇詭異巧涧,居然都是意外死亡,警方通過查閱死者的電腦和手機(jī)遥倦,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 88,644評(píng)論 2 381
  • 文/潘曉璐 我一進(jìn)店門谤绳,熙熙樓的掌柜王于貴愁眉苦臉地迎上來,“玉大人袒哥,你說我怎么就攤上這事缩筛。” “怎么了堡称?”我有些...
    開封第一講書人閱讀 153,340評(píng)論 0 344
  • 文/不壞的土叔 我叫張陵瞎抛,是天一觀的道長。 經(jīng)常有香客問我却紧,道長桐臊,這世上最難降的妖魔是什么胎撤? 我笑而不...
    開封第一講書人閱讀 55,449評(píng)論 1 279
  • 正文 為了忘掉前任,我火速辦了婚禮断凶,結(jié)果婚禮上伤提,老公的妹妹穿的比我還像新娘。我一直安慰自己认烁,他們只是感情好肿男,可當(dāng)我...
    茶點(diǎn)故事閱讀 64,445評(píng)論 5 374
  • 文/花漫 我一把揭開白布。 她就那樣靜靜地躺著却嗡,像睡著了一般舶沛。 火紅的嫁衣襯著肌膚如雪。 梳的紋絲不亂的頭發(fā)上稽穆,一...
    開封第一講書人閱讀 49,166評(píng)論 1 284
  • 那天冠王,我揣著相機(jī)與錄音,去河邊找鬼舌镶。 笑死柱彻,一個(gè)胖子當(dāng)著我的面吹牛,可吹牛的內(nèi)容都是我干的餐胀。 我是一名探鬼主播哟楷,決...
    沈念sama閱讀 38,442評(píng)論 3 401
  • 文/蒼蘭香墨 我猛地睜開眼,長吁一口氣:“原來是場(chǎng)噩夢(mèng)啊……” “哼否灾!你這毒婦竟也來了卖擅?” 一聲冷哼從身側(cè)響起,我...
    開封第一講書人閱讀 37,105評(píng)論 0 261
  • 序言:老撾萬榮一對(duì)情侶失蹤墨技,失蹤者是張志新(化名)和其女友劉穎惩阶,沒想到半個(gè)月后,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體扣汪,經(jīng)...
    沈念sama閱讀 43,601評(píng)論 1 300
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡断楷,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 36,066評(píng)論 2 325
  • 正文 我和宋清朗相戀三年,在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了崭别。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片冬筒。...
    茶點(diǎn)故事閱讀 38,161評(píng)論 1 334
  • 序言:一個(gè)原本活蹦亂跳的男人離奇死亡,死狀恐怖茅主,靈堂內(nèi)的尸體忽然破棺而出舞痰,到底是詐尸還是另有隱情,我是刑警寧澤诀姚,帶...
    沈念sama閱讀 33,792評(píng)論 4 323
  • 正文 年R本政府宣布响牛,位于F島的核電站,受9級(jí)特大地震影響,放射性物質(zhì)發(fā)生泄漏娃善。R本人自食惡果不足惜论衍,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 39,351評(píng)論 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一處隱蔽的房頂上張望聚磺。 院中可真熱鬧,春花似錦炬丸、人聲如沸瘫寝。這莊子的主人今日做“春日...
    開封第一講書人閱讀 30,352評(píng)論 0 19
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽焕阿。三九已至,卻和暖如春首启,著一層夾襖步出監(jiān)牢的瞬間暮屡,已是汗流浹背。 一陣腳步聲響...
    開封第一講書人閱讀 31,584評(píng)論 1 261
  • 我被黑心中介騙來泰國打工毅桃, 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留褒纲,地道東北人。 一個(gè)月前我還...
    沈念sama閱讀 45,618評(píng)論 2 355
  • 正文 我出身青樓钥飞,卻偏偏與公主長得像莺掠,于是被迫代替她去往敵國和親。 傳聞我的和親對(duì)象是個(gè)殘疾皇子读宙,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 42,916評(píng)論 2 344

推薦閱讀更多精彩內(nèi)容

  • 流是Node中最重要的組件和模式之一彻秆。在社區(qū)里有一句格言說:讓一切事務(wù)流動(dòng)起來。這已經(jīng)足夠來描述在Node中流...
    宮若石閱讀 543評(píng)論 0 0
  • Buffer Buffer的構(gòu)成 Buffer對(duì)象類似數(shù)組结闸,它的元素位16進(jìn)制的兩位數(shù)唇兑,即0到255的數(shù)值。主要是...
    人失格閱讀 1,799評(píng)論 0 0
  • 編譯地址:https://github.com/substack/stream-handbook譯者:jabez1...
    IT程序獅閱讀 1,293評(píng)論 0 4
  • Spring Cloud為開發(fā)人員提供了快速構(gòu)建分布式系統(tǒng)中一些常見模式的工具(例如配置管理桦锄,服務(wù)發(fā)現(xiàn)扎附,斷路器,智...
    卡卡羅2017閱讀 134,601評(píng)論 18 139
  • https://nodejs.org/api/documentation.html 工具模塊 Assert 測(cè)試 ...
    KeKeMars閱讀 6,305評(píng)論 0 6