Node.js 天生異步和事件驅(qū)動(dòng)，非常適合處理 I/O 相關(guān)的任務(wù)。如果你在處理應(yīng)用中 I/O 相關(guān)的操作，你可以利用 Node.js 中的流(stream)。因此，我們先具體看看流，理解一下它們是怎么簡(jiǎn)化 I/O 操作的吧。

流是什么

流是 unix 管道，讓你可以很容易地從數(shù)據(jù)源讀取數(shù)據(jù)，然后流向另一個(gè)目的地。

簡(jiǎn)單來(lái)說(shuō)，流不是什么特別的東西，它只是一個(gè)實(shí)現(xiàn)了一些方法的 EventEmitter 。根據(jù)它實(shí)現(xiàn)的方法，流可以變成可讀流(Readable)，可寫(xiě)流(Writable)，或者雙向流(Duplex，同時(shí)可讀可寫(xiě))。

可讀流能讓你從一個(gè)數(shù)據(jù)源讀取數(shù)據(jù)，而可寫(xiě)流則可以讓你往目的地寫(xiě)入數(shù)據(jù)。

如果你已經(jīng)用過(guò) Node.js，你很可能已經(jīng)遇到過(guò)流了。

例如，在一個(gè) Node.js 的 HTTP 服務(wù)器里面， request 是一個(gè)可讀流， response 是一個(gè)可寫(xiě)流。

你也可能用過(guò) fs 模塊，它能幫你處理可讀可寫(xiě)流。

現(xiàn)在讓你學(xué)一些基礎(chǔ)，理解不同類(lèi)型的流。本文會(huì)討論可讀流和可寫(xiě)流，雙向流超出了本文的討論范圍，我們不作討論。

可讀流 (Readable Streams)

我們可以用可讀流從一個(gè)數(shù)據(jù)源中讀取數(shù)據(jù)，這個(gè)數(shù)據(jù)源可以是任何東西，例如系統(tǒng)中的一個(gè)文件，內(nèi)存中的 buffer，甚至是其他流。因?yàn)榱魇?EventEmitter ，它們會(huì)用各種事件發(fā)送數(shù)據(jù)。我們會(huì)利用這些事件來(lái)讓流工作。

從流中讀取數(shù)據(jù)

從流中讀取數(shù)據(jù)最好的方式是監(jiān)聽(tīng) data 事件，添加一個(gè)回調(diào)函數(shù)。當(dāng)有數(shù)據(jù)流過(guò)來(lái)的時(shí)候，可讀流會(huì)發(fā)送 data 事件，回調(diào)函數(shù)就會(huì)觸發(fā)。看看下面的代碼片段：

var fs = require('fs');
var readableStream = fs.createReadStream('file.txt');
var data = '';

var readableStream.on('data', function(chunk){
 data += chunk;
});

readableStream.on('end', function(){
 console.log(data);
});

fs.createReadStream 會(huì)給你一個(gè)可讀流。

最開(kāi)始的時(shí)候，這個(gè)流不是流動(dòng)態(tài)的。當(dāng)你添加了 data 的事件監(jiān)聽(tīng)器，加上一個(gè)回調(diào)函數(shù)時(shí)，它才會(huì)變成流動(dòng)態(tài)的。在這之后，它就會(huì)讀取一小塊數(shù)據(jù)，然后傳到你的回調(diào)函數(shù)里面。

流的實(shí)現(xiàn)者決定了 data 事件的觸發(fā)頻率，例如 HTTP request 會(huì)在讀取到幾 KB 數(shù)據(jù)的時(shí)候觸發(fā) data 事件。 當(dāng)你從一個(gè)文件中讀取數(shù)據(jù)的時(shí)候，你可能會(huì)決定當(dāng)一行被讀完的時(shí)候就觸發(fā) data 事件。

當(dāng)沒(méi)有數(shù)據(jù)可讀的時(shí)候 (讀到文件尾部時(shí))，流就會(huì)發(fā)送 end 事件。在上面的例子中，我們監(jiān)聽(tīng)了這個(gè)事件，當(dāng)讀完文件的時(shí)候，就把數(shù)據(jù)打印出來(lái)。

還有另一種讀取流的方式，你只要在讀到文件尾部前不斷調(diào)用流實(shí)例中的 read() 方法就可以了。

var fs = require('fs');
var readableStream = fs.createReadStream('file.txt');
var data = '';
var chunk;

readableStream.on('readable', function(){
 while ((chunk = readableStream.read()) != null) {
 data += chunk;
 }
});

readableStream.on('end', function(){
 console.log(data);
});

read() 方法會(huì)從內(nèi)部 buffer 中讀取數(shù)據(jù)，當(dāng)沒(méi)有數(shù)據(jù)可讀的時(shí)候，它會(huì)返回 null 。

因此，在 while 循環(huán)中我們檢查 read() 是不是返回 null ，當(dāng)它返回 null 的時(shí)候，就終止循環(huán)。

需要注意的是，當(dāng)我們可以從流中讀取數(shù)據(jù)的時(shí)候， readable 事件就會(huì)觸發(fā)。

設(shè)置編碼

默認(rèn)情況下，你從流中讀取到的是 Buffer 對(duì)象。如果你要讀取的是字符串的話，這并不適合你。因此，你可以像下面的例子那樣通過(guò)調(diào)用 Readable.setEncoding() 來(lái)設(shè)置流的編碼：

var fs = require('fs');
var readableStream = fs.createReadStream('file.txt');
var data = '';

readableStream.setEncoding('utf8');

readableStream.on('data', function(chunk){
 data += chunk;
});

readableStream.on('end', function(){
 console.log(data);
});

上面的例子中，我們把流的編碼設(shè)置成 utf8 ，數(shù)據(jù)就會(huì)被解析成 utf8 ，回調(diào)函數(shù)中的 chunk 就會(huì)是字符串了。

管道 (Piping)

管道是一個(gè)很棒的機(jī)制，你不需要自己管理流的狀態(tài)就可以從數(shù)據(jù)源中讀取數(shù)據(jù)，然后寫(xiě)入到目的地中。我們先看看下面的例子：

var fs = require('fs');
var readableStream = fs.createReadStream('file1.txt');
var writableStream = fs.createWriteStream('file2.txt');

readableStream.pipe(writableStream);

上面的例子利用 pipe() 方法把 file1 的內(nèi)容寫(xiě)到 file2 中。因?yàn)?pipe() 會(huì)幫你管理數(shù)據(jù)流，你不需要擔(dān)心數(shù)據(jù)流的速度。這讓 pipe() 變得非常簡(jiǎn)潔易用。

需要注意的是， pipe() 會(huì)返回目的地的流，因此你可以很輕易讓多個(gè)流鏈接起來(lái)！

鏈接 (Chaining)

假設(shè)有一個(gè)歸檔文件，你想要解壓它。有很多方式可以完成這個(gè)任務(wù)。但最簡(jiǎn)潔的方式是利用管道和鏈接：

var fs = require('fs');
var zlib = require('zlib');

fs.createReadStream('input.txt.gz')
 .pipe(zlib.createGunzip())
 .pipe(fs.createWriteStream('output.txt'));

首先，我們通過(guò) input.txt.gz 創(chuàng)建了一個(gè)可讀流，然后讓它流 zlib.createGunzip() 流，它會(huì)解壓內(nèi)容。最后，我們添加一個(gè)可寫(xiě)流把解壓后的內(nèi)容寫(xiě)到另一個(gè)文件中。

其他方法

我們已經(jīng)討論了一些可讀流中重要的概念了，這里還有一些你需要知道的方法：

• Readable.pause() – 這個(gè)方法會(huì)暫停流的流動(dòng)。換句話說(shuō)就是它不會(huì)再觸發(fā) data 事件。
• Readable.resume() – 這個(gè)方法和上面的相反，會(huì)讓暫停流恢復(fù)流動(dòng)。
• Readable.unpipe() – 這個(gè)方法會(huì)把目的地移除。如果有參數(shù)傳入，它會(huì)讓可讀流停止劉翔某個(gè)特定的目的地，否則，它會(huì)移除所有目的地。

可寫(xiě)流 (Writable Streams)

可寫(xiě)流讓你把數(shù)據(jù)寫(xiě)入目的地。就像可讀流那樣，這些也是 EventEmitter ，它們也會(huì)觸發(fā)不同的事件。我們來(lái)看看可寫(xiě)流中會(huì)觸發(fā)的事件和方法吧。

寫(xiě)入流

要把數(shù)據(jù)寫(xiě)如到可寫(xiě)流中，你需要在可寫(xiě)流實(shí)例中調(diào)用 write() 方法，看看下面的例子：

var fs = require('fs');
var readableStream = fs.createReadStream('file1.txt');
var writableStream = fs.createWriteStream('file2.txt');

readableStream.setEncoding('utf8');

readableStream.on('data', function(chunk){
 writableStream.write('chunk');
});

上面的代碼非常簡(jiǎn)單，它只是從輸入流中讀取數(shù)據(jù)，然后用 write() 寫(xiě)入到目的地中。

這個(gè)方法返回一個(gè)布爾值來(lái)表示寫(xiě)入是否成功。如果返回的是 true 那表示寫(xiě)入成功，你可以繼續(xù)寫(xiě)入更多的數(shù)據(jù)。 如果是 false ，那意味著發(fā)生了什么錯(cuò)誤，你現(xiàn)在不能繼續(xù)寫(xiě)入了。可寫(xiě)流會(huì)觸發(fā)一個(gè) drain 事件來(lái)告訴你你可以繼續(xù)寫(xiě)入數(shù)據(jù)。

寫(xiě)完數(shù)據(jù)后

當(dāng)你不需要在寫(xiě)入數(shù)據(jù)的時(shí)候，你可以調(diào)用 end() 方法來(lái)告訴流你已經(jīng)完成寫(xiě)入了。假設(shè) res 是一個(gè) HTTP response 對(duì)象，你通常會(huì)發(fā)送響應(yīng)給瀏覽器：

res.write('Some Data!!');
res.end();

當(dāng) end() 被調(diào)用時(shí)，所有數(shù)據(jù)會(huì)被寫(xiě)入，然后流會(huì)觸發(fā)一個(gè) finish 事件。注意在調(diào)用 end() 之后，你就不能再往可寫(xiě)流中寫(xiě)入數(shù)據(jù)了。例如下面的代碼就會(huì)報(bào)錯(cuò)：

res.write('Some Data!!');
res.end();
res.write('Trying to write again'); //Error !

這里有一些和可寫(xiě)流相關(guān)的重要事件：

• error – 在寫(xiě)入或鏈接發(fā)生錯(cuò)誤時(shí)觸發(fā)
• pipe – 當(dāng)可讀流鏈接到可寫(xiě)流時(shí)，這個(gè)事件會(huì)觸發(fā)
• unpipe – 在可讀流調(diào)用 unpipe 時(shí)會(huì)觸發(fā)

更多精彩內(nèi)容和產(chǎn)品推薦，請(qǐng)咨詢！