為什么80%的碼農(nóng)都做不了架構(gòu)師？>>> ??

stream

流是一個(gè)抽象接口，在 Node 里被不同的對(duì)象實(shí)現(xiàn)。例如?request to an HTTP server?是流，stdout?是流。流是可讀，可寫(xiě)，或者可讀寫(xiě)。所有的流是?EventEmitter?的實(shí)例。

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

這個(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），可寫(xiě)（Writable），或者兼具兩者（Duplex，雙工）的。

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

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

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

基本所有的 Node 程序，無(wú)論多簡(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 Streamvar body = '';// we want to get the data as utf8 strings// If you don't set an encoding, then you'll get Buffer objectsreq.setEncoding('utf8');// 可讀流（Readable stream） emit 'data' 事件 once a 監(jiān)聽(tīng)器（listener） is addedreq.on('data', function (chunk) {body += chunk;});// the end 事件 tells you that you have entire bodyreq.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

類(lèi)： stream.Readable

可讀流（Readable stream）接口是對(duì)你正在讀取的數(shù)據(jù)的來(lái)源的抽象。換句話說(shuō)，數(shù)據(jù)來(lái)來(lái)自

可讀流（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() 來(lái)讀取數(shù)據(jù)。 暫停模式（paused mode） 是默認(rèn)模式。

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

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

添加一個(gè)?'data'?事件 事件處理器來(lái)監(jiān)聽(tīng)數(shù)據(jù).
調(diào)用?resume()?方法來(lái)明確的開(kāi)啟數(shù)據(jù)流。
調(diào)用?pipe()?方法來(lái)發(fā)送數(shù)據(jù)給Writable.
可以通過(guò)以下方法來(lái)切換到暫停模式（paused mode）:

如果沒(méi)有 導(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)聽(tīng)器并不會(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' 事件.`

某些情況下, 如果沒(méi)有準(zhǔn)備好，監(jiān)聽(tīng)一個(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)聽(tīng)器（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'

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

注意，除非數(shù)據(jù)已經(jīng)被完全消費(fèi)， the end 事件才會(huì)觸發(fā)。 可以通過(guò)切換到流動(dòng)模式（flowing mode）來(lái)實(shí)現(xiàn)，或者通過(guò)調(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ù)。如果沒(méi)有可用數(shù)據(jù)，將會(huì)返回null
如果傳了 size參數(shù)，將會(huì)返回相當(dāng)字節(jié)的數(shù)據(jù)。如果size不可用，將會(huì)返回 null
如果你沒(méi)有指定 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()][] 來(lái)打開(kāi)數(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} 寫(xiě)入數(shù)據(jù)的目標(biāo)
• options?{Object} 導(dǎo)流（pipe） 選項(xiàng)
• end?{Boolean} 讀取到結(jié)束符時(shí)，結(jié)束寫(xiě)入者。默認(rèn) =?true

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

可以導(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?不可再寫(xiě)。傳?{ end:false}作為options，可以保持目標(biāo)流打開(kāi)狀態(tài)。

這會(huì)讓?writer保持打開(kāi)狀態(tài)，可以在最后寫(xiě)入"Goodbye" 。

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

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

readable.unpipe([destination])

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

這個(gè)方法會(huì)解除之前調(diào)用?pipe()?設(shè)置的鉤子（?pipe()?）。
如果沒(méi)有指定?destination,所有的 導(dǎo)流（pipe） 都會(huì)被移除。
如果指定了?destination，但是沒(méi)有建立如果沒(méi)有指定?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)”，返回到原來(lái)的源，以便流能將它傳遞給其它消費(fèi)者。

如果你在程序中必須經(jīng)常調(diào)用?stream.unshift(chunk)?，那你可以考慮實(shí)現(xiàn)?Transform?來(lái)替換（參見(jiàn)下文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àn)下文“兼容性”章節(jié)）。

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

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

例如:

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

類(lèi)： stream.Writable

<!--type=class-->

可寫(xiě)流（Writable stream ）接口是你正把數(shù)據(jù)寫(xiě)到一個(gè)目標(biāo)的抽象。
可寫(xiě)流（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} 準(zhǔn)備寫(xiě)的數(shù)據(jù)
• encoding?{String} 編碼方式（如果chunk?是字符串）
• callback?{Function} 數(shù)據(jù)塊寫(xiě)入后的回調(diào)
• 返回: {Boolean} 如果數(shù)據(jù)已被全部處理返回true

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

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

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

事件: 'drain'

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

// Write the data to the supplied 可寫(xiě)流（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)制緩存所有寫(xiě)入。

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

writable.uncork()

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

writable.setDefaultEncoding(encoding)

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

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

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

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

當(dāng)沒(méi)有更多的數(shù)據(jù)寫(xiě)入的時(shí)候調(diào)用這個(gè)方法。如果給出，回調(diào)會(huì)被用作 finish 事件的監(jiān)聽(tīng)器。

調(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)寫(xiě)入到底層系統(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）到可寫(xiě)流的源流

無(wú)論何時(shí)在可寫(xiě)流（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

無(wú)論何時(shí)在可寫(xiě)流（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}

寫(xiě)或?qū)Я?#xff08;pipe）數(shù)據(jù)時(shí)，如果有錯(cuò)誤會(huì)觸發(fā)。

類(lèi)： stream.Duplex

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

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

• tcp sockets
• zlib streams
• crypto streams

類(lèi)： stream.Transform

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

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

• zlib streams
• crypto streams

API for Stream Implementors

<!--type=misc-->

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

在你的子類(lèi)中擴(kuò)展適合的父類(lèi). (util.inherits?方法很有幫助)

在你的構(gòu)造函數(shù)中調(diào)用父類(lèi)的構(gòu)造函數(shù)，以確保內(nèi)部的機(jī)制初始化正確。

實(shí)現(xiàn)一個(gè)或多個(gè)方法，如下所列

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

<table>
<thead>
<tr>
<th>

Use-case

</th>
<th>

Class

</th>
<th>

方法(s) to implement

</th>
</tr>
</thead>
<tr>
<td>

Reading only

</td>
<td>

Readable

</td>
<td>

_read

</td>
</tr>
<tr>
<td>

Writing only

</td>
<td>

Writable

</td>
<td>

_write

</td>
</tr>
<tr>
<td>

Reading and writing

</td>
<td>

Duplex

</td>
<td>

_read,?_write

</td>
</tr>
<tr>
<td>

Operate on written data, then read the result

</td>
<td>

Transform

</td>
<td>

_transform,?_flush

</td>
</tr>
</table>

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

類(lèi)： stream.Readable

<!--type=class-->

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

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

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

<!--type=example-->

這是可讀流（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ù)類(lèi)似, 但它被實(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\nvar 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ì)象的流。意思是說(shuō) stream.read(n) 返回一個(gè)單獨(dú)的值，而不是大小為 n 的 Buffer。

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

readable._read(size)

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

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

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

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

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

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

size?參數(shù)僅供參考. 調(diào)用 “read” 可以知道知道應(yīng)當(dāng)抓取多少數(shù)據(jù)；其余與之無(wú)關(guān)的實(shí)現(xiàn)，比如 TCP 或 TLS，則可忽略這個(gè)參數(shù)，并在可用時(shí)返回?cái)?shù)據(jù)。例如，沒(méi)有必要“等到” 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?類(lèi)將數(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ì)成盡可能地靈活。比如說(shuō)，你可以包裝一個(gè)低級(jí)別的，具備某種暫停/恢復(fù)機(jī)制，和數(shù)據(jù)回調(diào)的數(shù)據(jù)源。這種情況下，你可以通過(guò)這種方式包裝低級(jí)別來(lái)源對(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(); };

類(lèi)： stream.Writable

<!--type=class-->

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

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

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，可以寫(xiě)任意數(shù)據(jù)，而不僅僅是Buffer?/?String. 默認(rèn)=?false

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

writable._write(chunk, encoding, callback)

• chunk?{Buffer | String} 要寫(xiě)入的數(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ù))。

所以可寫(xiě)流（Writable stream ） 實(shí)現(xiàn)必須提供一個(gè)?_write()方法，來(lái)發(fā)送數(shù)據(jù)給底層資源。
注意:?這個(gè)函數(shù)不能直接調(diào)用?,由子類(lèi)實(shí)現(xiàn)， 僅內(nèi)部可寫(xiě)方法可以調(diào)用。
使用標(biāo)準(zhǔn)的?callback(error)?方法調(diào)用回調(diào)函數(shù)，來(lái)表明寫(xiě)入完成或遇到錯(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)。如果你沒(méi)有明確的設(shè)置decodeStrings?為?false，這樣你就可以安不管?encoding?參數(shù)，并假定?chunk?一直是一個(gè)緩存。

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

writable._writev(chunks, callback)

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

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

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

類(lèi)：stream.Duplex

<!--type=class-->

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

由于 JavaScript 并沒(méi)有多重繼承能力，因此這個(gè)類(lèi)繼承自 Readable，寄生自 Writable.從而讓用戶在雙工擴(kuò)展類(lèi)中同時(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)寫(xiě)端結(jié)束的時(shí)候，流會(huì)自動(dòng)的結(jié)束讀端，反之亦然。
• readableObjectMode?{Boolean} 默認(rèn)=false. 將?objectMode?設(shè)為讀端的流，如果為?true，將沒(méi)有效果。
• writableObjectMode?{Boolean} 默認(rèn)=false. 將?objectMode設(shè)為寫(xiě)端的流，如果為?true，將沒(méi)有效果。

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

類(lèi)：stream.Transform

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

輸入輸出沒(méi)有要求大小相同，塊數(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()?方法（參見(jiàn)如下）。

new stream.Transform([options])

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

擴(kuò)展自 轉(zhuǎn)換流（transform class) 的類(lèi)，確保調(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)用。?由子類(lèi)實(shí)現(xiàn)，僅內(nèi)部可寫(xiě)方法可以調(diào)用.

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

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

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

僅在當(dāng)前數(shù)據(jù)塊完全消費(fèi)后調(diào)用這個(gè)回調(diào)。注意，輸入塊可能有，也可能沒(méi)有對(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); }

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

transform._flush(callback)

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

注意:?這個(gè)函數(shù)不能直接調(diào)用。?由子類(lèi)實(shí)現(xiàn)，僅內(nèi)部可寫(xiě)方法可以調(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)所有的寫(xiě)入數(shù)據(jù)被消費(fèi)后，分發(fā)end告訴讀端。和?_transform?一樣，當(dāng)刷新操作完畢，?transform.push(chunk)?為0或更多次數(shù)，。

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

事件: 'finish' and 'end'

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

Example:?SimpleProtocol?parser v2

上面的簡(jiǎn)單協(xié)議分析例子列子可以通過(guò)使用高級(jí)別的[Transform][] 流來(lái)實(shí)現(xiàn)，和?parseHeader?，?SimpleProtocol v1列子類(lèi)似。

在這個(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.

類(lèi)：stream.PassThrough

這是 Transform 流的簡(jiǎn)單實(shí)現(xiàn)，將輸入的字節(jié)簡(jiǎn)單的傳遞給輸出。它的主要用途是測(cè)試和演示。偶爾要構(gòu)建某種特殊流時(shí)也會(huì)用到。

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

<!--type=misc-->

緩沖

<!--type=misc-->

可寫(xiě)流（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ù)到可寫(xiě)流（Writable stream）。即使?write()?返回?false。

流（尤其是pipe()?方法）得目的是限制數(shù)據(jù)的緩存量到一個(gè)可接受的水平，使得不同速度的源和目的不會(huì)淹沒(méi)可用內(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)用。

雖然基本上沒(méi)有必要這么做。但你在 Node 內(nèi)部的某些地方看到它確實(shí)這么做了，尤其是在 Readable 流類(lèi)的內(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)程. 然而，它沒(méi)有添加任何數(shù)據(jù)到可讀緩沖區(qū)中，所以沒(méi)有東西可供用戶消費(fèi)。

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

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

和老版本的兼容性

<!--type=misc-->

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

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

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

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

• 沒(méi)有添加?'data'?事件 處理器
• 從來(lái)沒(méi)有調(diào)用?resume()?方法
• 流從來(lái)沒(méi)有被倒流（pipe）到任何可寫(xiě)目標(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()?方法來(lái)開(kāi)始工作：

// 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è)可讀類(lèi)里。

Object Mode

<!--type=misc-->

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

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

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

在對(duì)象模式里， 可寫(xiě)流（Writable stream ） 總會(huì)忽略傳給stream.write(data, encoding)的?encoding參數(shù)。

特殊值?null?在對(duì)象模式里，依舊保持它的特殊性。也就說(shuō)，對(duì)于對(duì)象模式的可讀流（Readable stream），stream.read()?返回 null 意味著沒(méi)有更多數(shù)據(jù)，同時(shí)stream.push(null)?會(huì)告知流數(shù)據(jù)結(jié)束（EOF）。

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

應(yīng)該在你的子類(lèi)構(gòu)造函數(shù)里，設(shè)置objectMode?。在過(guò)程中設(shè)置不安全。

對(duì)于雙工流（Duplex streams），objectMode可以用readableObjectMode?和?writableObjectMode?分別為讀寫(xiě)端分別設(shè)置。這些選項(xiàng)，被轉(zhuǎn)換流（Transform streams）用來(lái)實(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(); };

轉(zhuǎn)載于:https://my.oschina.net/fymoon/blog/796488

總結(jié)

以上是生活随笔為你收集整理的node.js之stream模块的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。