摘要：當(dāng)一個客戶端的響應(yīng)對象是一個可讀流，那么在服務(wù)器端這就是一個可寫流。的模塊給我們提供了一個可以操作任何文件的可讀流通過方法創(chuàng)建。創(chuàng)建一個可讀流創(chuàng)建可讀流，我們需要類創(chuàng)建一個可讀流非常簡單?？梢酝ㄟ^修改可讀流配置里面的方法實(shí)現(xiàn)。

Node.js的stream模塊是有名的應(yīng)用困難，更別說理解了。那現(xiàn)在可以告訴你，這些都不是問題了。

多年來，開發(fā)人員在那里創(chuàng)建了大量的軟件包，其唯一目的就是使用stream使用起來更簡單，但是在這篇文章里，我們專注于介紹原生的Node.js Steam Api。

"Stream 是Node.js中最好的卻最容易被誤解的部分" ----- Dominic Tarr

Streams是數(shù)據(jù)的集合，就跟數(shù)組和字符串一樣。不同點(diǎn)就在于Streams可能不是立刻就全部可用，并且不會全部載入內(nèi)存。這使得他非常適合處理大量數(shù)據(jù)，或者處理每隔一段時間有一個數(shù)據(jù)片段傳入的情況。

但是，Stream并不僅僅適用于處理大數(shù)據(jù)(大塊的數(shù)據(jù)。。。)。使用它，同樣也有利于組織我們大代碼。就像我們使用管道去和合并強(qiáng)大的Linux命令。在Node.js中，我們也可以做同樣的事情。

const grep = ... // A stream for the grep output
const wc = ... // A stream for the wc input
grep.pipe(wc)

Node.js的很多內(nèi)置模塊都實(shí)現(xiàn)了Stream接口

上面例子里面的Node.js對象列表包括了可讀流和可寫流，有一些對象既是可讀流也是可寫流，像TCP sockets, zlib 和 crypto streams。

注意這些對象是有很密切的關(guān)聯(lián)的。當(dāng)一個客戶端的HTTP 響應(yīng)對象是一個可讀流，那么在服務(wù)器端這就是一個可寫流。因?yàn)樵贖TTP例子中，我們通常是從一個對象(http.IncomingMessage)讀取再寫入到另外一個對象(http.ServerResponse)中去。

還要注意，當(dāng)涉及到子進(jìn)程時，stdio流（stdin，stdout，stderr）具有逆流類型。這就允許我們非常方便的使用管道從主進(jìn)程連接子進(jìn)程的Streams。

理論都是很好的，但事實(shí)到底是怎么樣子的呢？讓我們看一些例子示范代碼Streams在內(nèi)存使用方面的比較。

我們先創(chuàng)建一個大文件

const fs = require("fs");
const file = fs.createWriteStream("./big.file");

for(let i=0; i<= 1e6; i++) {
  file.write("Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.
");
}

file.end();

看看我使用什么創(chuàng)建文件的？一個可寫流嘛

fs模塊可以通過Stream接口來讀取和寫入文件。在上面的例子中，我們在循環(huán)中通過可寫流向big.file寫入了1百萬行數(shù)據(jù)。

運(yùn)行上面的代碼會生成一個大概400M的文件

這是一個簡單的Node web服務(wù)器，專門為big.file提供服務(wù)：

const fs = require("fs");
const server = require("http").createServer();

server.on("request", (req, res) => {
  fs.readFile("./big.file", (err, data) => {
    if (err) throw err;
  
    res.end(data);
  });
});

server.listen(8000);

當(dāng)server收到請求，它會使用異步方法fs.readFile處理這個big file。但是這并不代表我們會打斷事件循環(huán)機(jī)制。一切都是正確的嗎？？

那現(xiàn)在當(dāng)我們啟動server，看看內(nèi)存監(jiān)視器都發(fā)生了什么。

現(xiàn)在訪問這個服務(wù)器，看看內(nèi)存的使用情況。

內(nèi)存占用立刻飆升到434.8 MB。

在我們把文件內(nèi)容輸出到客戶端之前，我們就把整個文件讀入了內(nèi)存。這是很低效的。

HTTP response對象(上文中的res對象)也是一個可寫流，這就意味著如果我們有一個代表著big file的可讀流，我們可以通過管道把他們倆連接起來實(shí)現(xiàn)同樣的功能，而不需要使用400M內(nèi)存。

Node的fs模塊給我們提供了一個可以操作任何文件的可讀流,通過createReadStream方法創(chuàng)建。我們可以把它和response對象連接起來。

const fs = require("fs");
const server = require("http").createServer();

server.on("request", (req, res) => {
  const src = fs.createReadStream("./big.file");
  src.pipe(res);
});

server.listen(8000);

現(xiàn)在再去訪問server的時候，令人驚訝的事情發(fā)生了(看內(nèi)存監(jiān)視器)

發(fā)生了什么？

當(dāng)我們訪問服務(wù)器的時候，我們通過流每次使用一段數(shù)據(jù)，這意味著我們不是把全部的數(shù)據(jù)都加載到內(nèi)存中。內(nèi)存使用量只上升了不到25M。

可以把上面的例子用到極致，生成5百萬行數(shù)據(jù)而不是1百萬行。這樣子的話，這個文件的大小會超過2GB，這實(shí)際上大于Node中的默認(rèn)緩沖區(qū)限制。

如果你想在server上使用fs.readFile,這在默認(rèn)情況下是行不通的，除非你改了Node.js的默認(rèn)緩沖區(qū)限制。但是使用fs.createReadStream，把2 GB的數(shù)據(jù)返回給客戶端根本不存在問題，甚至內(nèi)存使用量都沒有任何變化。

準(zhǔn)備好學(xué)習(xí)Steam了嗎？

在Node.js中有4中基本的流類型：Readable, Writable, Duplex, and Transform streams。

Readable 可讀流是可以從中消耗數(shù)據(jù)的源的抽象，一個例子就是fs.createReadStream方法

Writable 可寫流是可以寫入數(shù)據(jù)的目標(biāo)的抽象，一個例子就是fs.createWriteStream方法

duplex Steam是一個同時具有讀寫功能的流，一個例子就是TCP socket

Transform 是一個雙工流，它可以在交換數(shù)據(jù)的時候做轉(zhuǎn)換。一個例子就是zlib.createGzip使用gzip壓縮數(shù)據(jù)。你可以把Transform streams當(dāng)成是一個傳入可讀流，返回一個可寫流的函數(shù)。它還有一個別名through streams

所有的Stream都是EventEmitter的實(shí)例對象。當(dāng)流讀和寫的時候都會觸發(fā)相應(yīng)的事件。但是還有一個更簡單的使用方法，那就是使用pipe。

要記住下面這個魔幻方法

readableSrc.pipe(writableDest)

在這一行里面，我們通過管道把可讀流(源)輸出到一個可寫流里面去(目標(biāo))，源必須是一個可寫流，目標(biāo)必須是可寫流。當(dāng)然，他們也都可以是duplex/Transform。事實(shí)上，當(dāng)我們使用管道連接流的時候，我們可以像在linux中一樣使用鏈?zhǔn)竭B接。

readableSrc
  .pipe(transformStream1)
  .pipe(transformStream2)
  .pipe(finalWrtitableDest)

pipe方法返回目標(biāo)流，這保證了我們可以使用鏈?zhǔn)秸{(diào)用。對于streams a(可讀流)，b,c(可讀可寫流)，d可寫流，我們可以使用：

a.pipe(b).pipe(c).pipe(d)
# Which is equivalent to:
a.pipe(b)
b.pipe(c)
c.pipe(d)
# Which, in Linux, is equivalent to:
$ a | b | c | d

pipe方法是使用流最簡便的方法。通常通過管道和事件的方法使用流，但是要盡量避免兩者混用。通常當(dāng)你使用pipe方法就不需要使用事件了。但是當(dāng)你需要更多定制的操作的話，使用事件的方式會更好。

除了從可讀流讀取數(shù)據(jù)傳輸?shù)娇蓪懥鳎?b>pipe方法還自動處理一些其他事情。比如處理錯誤，處理文件結(jié)束操作，流之間速度快慢問題。

同時，流也可以直接使用事件操作。以下是和管道相等的通過事件操作流的方法。

# readable.pipe(writable)

readable.on("data", (chunk) => {
  writable.write(chunk);
});

readable.on("end", () => {
  writable.end();
});

下面是一些重要流的事件和方法。

這些事件和方法在某種程度上是相關(guān)的，因?yàn)樗鼈兺ǔ１灰黄鹗褂谩?/p>

可讀流上的最重要的事件是

data事件，當(dāng)可讀流傳輸了一段數(shù)據(jù)的時候會觸發(fā)

end事件，當(dāng)沒有數(shù)據(jù)被傳輸時觸發(fā)

可寫流上的最重要的事件是

drain事件，當(dāng)可寫流可以接收事件的時候被觸發(fā)

finish事件，當(dāng)所有數(shù)據(jù)被接收時被觸發(fā)

事件和方法可以結(jié)合起來，以便定制和優(yōu)化流的使用。讀取可讀流，我們可以使用pipe/unpipe方法，或者read/unshift/resume方法。使用可寫流，我們可以可寫流作為pipe/unpipe方法的參數(shù)，或者使用write方法寫入，使用end方法關(guān)閉。

可讀流有兩個很重要的模式影響了我們使用的方式。

暫停模式

流動模式

這些模式有時候被稱為拉和推模式

所有的可讀流開始的時候都是默認(rèn)暫停模式，但是它們可以輕易的被切換成流動模式，當(dāng)我們需要的時候又可以切換成暫停模式。有時候這個切換是自動的。

當(dāng)一個可讀流是暫停模式的時候，我們可以使用read方法從流中讀取。但是當(dāng)一個流是流動模式的時候，數(shù)據(jù)是持續(xù)的流動，我們需要使用事件去監(jiān)聽數(shù)據(jù)的變化。

在流動模式中，如果可讀流沒有監(jiān)聽者，可讀流的數(shù)據(jù)會丟失。這就是為什么當(dāng)可讀流逝流動模式的時候，我們必須使用data事件去監(jiān)聽數(shù)據(jù)的變化。事實(shí)上，只需添加一個數(shù)據(jù)事件處理程序即可將暫停的流轉(zhuǎn)換為流模式，刪除數(shù)據(jù)事件處理程序?qū)⒘髑袚Q回暫停模式。 其中一些是為了與舊的Node Stream接口進(jìn)行向后兼容。

可以使用resume()和pause()方法在這兩種模式之間切換。

當(dāng)我們使用pipe方法操作可讀流的時候是不需要擔(dān)心上面的這些操作的，因?yàn)?b>pipe方法會自動幫我們處理這些問題。

當(dāng)我們討論Node.js中的流時，有兩項(xiàng)重要的任務(wù)：

流的創(chuàng)建

流的使用

我們到現(xiàn)在為止討論的都是如何使用流，那下面來看看如何創(chuàng)建吧！

Streams的創(chuàng)建通常使用stream模塊。

為了創(chuàng)建一個可寫流，我們需要使用stream模塊里面的Writable類。

const { Writable } = require("stream");

我們可以有很多種方式實(shí)現(xiàn)一個可寫流。例如，我們可以繼承Writable類。

class myWritableStream extends Writable {

}

但是我更喜歡使用構(gòu)造函數(shù)的方式創(chuàng)建。通過給Writable傳遞一些參數(shù)來創(chuàng)建一個對象。唯一必須要傳的選項(xiàng)時write方法，它需要暴漏需要寫入的數(shù)據(jù)塊。

const { Writable } = require("stream");
const outStream = new Writable({
  write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();
  }
});

process.stdin.pipe(outStream);

write方法接收3個參數(shù)

chunk通常是一個buffer對象，我們可以通過配置修改

encoding在這種情況下就需要了，不過通常情況是可以忽略的

callback是當(dāng)我們處理完這個數(shù)據(jù)塊的時候需要調(diào)用的函數(shù)。這是一個寫入是否成功的信號。如果失敗了，給這個回調(diào)傳遞一個Error對象

在outStream中，我們簡單的把chunk打印出來，因?yàn)椴]有發(fā)生錯誤，我們直接調(diào)用了callback方法。這是這是簡單并不實(shí)用的打印流。它會打印接收到的所有值。

為了使用這個流，我們可以簡單的process.stdin這個可讀流。通過pipe方法連接起來。

當(dāng)我們運(yùn)行上面的例子，任何我們在控制臺輸入的內(nèi)容都會被console.log打印出來。

這不是一個非常實(shí)用的流的實(shí)現(xiàn)，但是它已經(jīng)被Node.js內(nèi)置實(shí)現(xiàn)了。outStream功能和process.stdout基本類似。我們也可以通過pipe方法把stdin和stdout連接起來并實(shí)現(xiàn)同樣的功能。

process.stdin.pipe(process.stdout);

創(chuàng)建可讀流，我們需要Readable類

const { Readable } = require("stream");
const inStream = new Readable({});

創(chuàng)建一個可讀流非常簡單?？梢允褂?b>push方法推入數(shù)據(jù)給其他流使用

const { Readable } = require("stream"); 
const inStream = new Readable();
inStream.push("ABCDEFGHIJKLM");
inStream.push("NOPQRSTUVWXYZ");
inStream.push(null); // No more data
inStream.pipe(process.stdout);

當(dāng)我們push一個null對象進(jìn)去的時候，這就標(biāo)志著我們要終止傳輸了。

我們可以簡單的把這個流通過pipe方法連接到一個可寫流process.stdout

運(yùn)行上面的代碼，會獲取所有的inStream的數(shù)據(jù)并打印出來。非常簡單但有效。

我們在通過pipe連接之前，就會把所有的數(shù)據(jù)推送到流里面。更好的方法是在消費(fèi)者要求時按需推送數(shù)據(jù)?？梢酝ㄟ^修改可讀流配置里面的read()方法實(shí)現(xiàn)。

const inStream = new Readable({
  read(size) {
    // there is a demand on the data... Someone wants to read it.
  }
});

當(dāng)讀取方法在可讀流上被調(diào)用時，該實(shí)現(xiàn)可以將部分?jǐn)?shù)據(jù)推送到隊(duì)列。 例如，我們可以一次推一個字母，從字符代碼65（表示A）開始，并在每次推送時遞增：

const inStream = new Readable({
  read(size) {
    this.push(String.fromCharCode(this.currentCharCode++));
    if (this.currentCharCode > 90) {
      this.push(null);
    }
  }
});
inStream.currentCharCode = 65;
inStream.pipe(process.stdout);

當(dāng)有流在讀取可讀流的數(shù)據(jù)的時候，read方法會持續(xù)執(zhí)行，這樣就會一直推出更多的字符。我們需要在某個時刻終止它，這就是為什么我們設(shè)置了一個終止條件推入了null。

我們應(yīng)該始終按需推送數(shù)據(jù)。

使用Duplex流，我們通過同一個對象實(shí)現(xiàn)可讀流和可寫流。這類似同時實(shí)現(xiàn)了兩個接口。

下面這個例子就結(jié)合了上面兩個可讀流和可寫流的綜合例子。

const { Duplex } = require("stream");

const inoutStream = new Duplex({
  write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();
  },

  read(size) {
    this.push(String.fromCharCode(this.currentCharCode++));
    if (this.currentCharCode > 90) {
      this.push(null);
    }
  }
});

inoutStream.currentCharCode = 65;
process.stdin.pipe(inoutStream).pipe(process.stdout);

通過合并這些方法，我們可以使用這個duplex流讀取從A-Z的字母也同樣可以使用它的打印功能。我們把stdin流連接到這個duplex上去使用它的打印功能，再把這個duplex流本身連接到stdout上去就在控制臺看到了A-Z。

雙工流的可讀寫的兩側(cè)完全獨(dú)立運(yùn)行。就像一個對象上兩種獨(dú)立的功能。

transform流是一種更有趣的duplex流。因?yàn)樗妮敵鰜碓从谒妮斎搿?/p>

對于一個transform流，我們不需要實(shí)現(xiàn)read和write方法，我們僅僅需要實(shí)現(xiàn)transform方法，這個方法合并了它們兩個。它具有寫入方法的功能，也可以用它推送數(shù)據(jù)。

這是一個簡單的transform例子，把任何輸入轉(zhuǎn)換成大寫。

const { Transform } = require("stream");

const upperCaseTr = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

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

在這個transformstream里面，像上個例子中雙工流一樣。但是我們只實(shí)現(xiàn)了transform()方法。我們把chunk轉(zhuǎn)換成大寫，再把大寫字母作為可讀流的輸入。

默認(rèn)，流會接收 Buffer/String 類型的數(shù)據(jù)。還有個字段 objectMode 設(shè)置，可以讓stream 接收任意類型的對象。

下面是一個這種類型的例子。以下變換流的組合使得將逗號分隔值的字符串映射為JavaScript對象的功能。 所以“a，b，c，d”成為{a：b，c：d}。

const { Transform } = require("stream");

const commaSplitter = new Transform({
  readableObjectMode: true,
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().trim().split(","));
    callback();
  }
});

const arrayToObject = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    const obj = {};
    for(let i=0; i < chunk.length; i+=2) {
      obj[chunk[i]] = chunk[i+1];
    }
    this.push(obj);
    callback();
  }
});

const objectToString = new Transform({
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    this.push(JSON.stringify(chunk) + "
");
    callback();
  }
});

process.stdin
  .pipe(commaSplitter)
  .pipe(arrayToObject)
  .pipe(objectToString)
  .pipe(process.stdout)

我們通過commasplitter傳遞輸入字符串（例如，“a，b，c，d”），它將數(shù)組作為其可讀數(shù)據(jù)（[“a”，“b”，“c”，“d”]））。 在該流上添加可讀的ObjectMode標(biāo)志是必要的，因?yàn)槲覀冋趯ο笸扑偷狡渖?，而不是字符串?/p>

然后我們把數(shù)組導(dǎo)入到arrayToObject數(shù)據(jù)流中，我們需要把writableObjectMode設(shè)置為 true，以表示arrayToObject會接收一個對象。另外它還會推送一個對象出去，所以還要把他的readableObjectMode為true。最后一個objectToString接收一個對象但是輸出字符串，所以就只需要設(shè)置一個writableObjectMode。

Node有一些非常有用的內(nèi)置transform streams對象。這包括zlib和crypto。

下面這個例子使用了zlib.createGzip()結(jié)合了額fs readable/writable streams實(shí)現(xiàn)了文件壓縮。

const fs = require("fs");
const zlib = require("zlib");
const file = process.argv[2];

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream(file + ".gz"));

你可以使用上面的腳本壓縮任何你傳入的參數(shù)文件。我們把文件的可讀流傳入了zlib的內(nèi)置轉(zhuǎn)換流。再寫入到新的.gz文件中。

使用管道還有一個很酷的事情，就是可以和事件結(jié)合起來。比如我想用戶看到進(jìn)度，并在結(jié)束的時候發(fā)個消息。因?yàn)?b>pipe方法會返回目標(biāo)流，我們也可以通過鏈?zhǔn)阶允录?/p>

const fs = require("fs");
const zlib = require("zlib");
const file = process.argv[2];

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .on("data", () => process.stdout.write("."))
  .pipe(fs.createWriteStream(file + ".zz"))
  .on("finish", () => console.log("Done"));

所以使用管道方法，我們可以輕松地操作流，但是我們還可以使用需要的事件進(jìn)一步定制與這些流的交互。

管道方法的好處是，我們可以用它來以一種可讀的方式逐一構(gòu)成我們的程序。 例如，我們可以簡單地創(chuàng)建一個變換流來報(bào)告進(jìn)度，而不用監(jiān)聽上面的數(shù)據(jù)事件，并用另一個.pipe()調(diào)用替換 .on() 調(diào)用：

const fs = require("fs");
const zlib = require("zlib");
const file = process.argv[2];

const { Transform } = require("stream");

const reportProgress = new Transform({
  transform(chunk, encoding, callback) {
    process.stdout.write(".");
    callback(null, chunk);
  }
});

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file + ".zz"))
  .on("finish", () => console.log("Done"));

reportProgress流是一個簡單的pass-through流，但是也跟標(biāo)準(zhǔn)事件一樣報(bào)告進(jìn)度。注意callback()函數(shù)的第二個參數(shù)，這相當(dāng)于把數(shù)據(jù)推送出去。

結(jié)合流的應(yīng)用是無止境的。例如，如果我們需要在我們gzip之前或之后加密文件，我們需要做的就是按照我們需要的確切順序來管理另一個轉(zhuǎn)換流。使用Node的crypto模塊處理這個事情。

const crypto = require("crypto");
// ...

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(crypto.createCipher("aes192", "a_secret"))
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file + ".zz"))
  .on("finish", () => console.log("Done"));

上面的腳本壓縮然后加密傳遞的文件，只有具有密碼的人才可以使用文件。 我們無法使用正常的解壓縮實(shí)用程序解壓縮此文件，因?yàn)樗驯患用堋?/p>

為了能夠解壓縮文件，我們需要使用完全相反的操作，這也很簡單。

fs.createReadStream(file)
  .pipe(crypto.createDecipher("aes192", "a_secret"))
  .pipe(zlib.createGunzip())
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file.slice(0, -3)))
  .on("finish", () => console.log("Done"));

假設(shè)傳遞的文件是壓縮版本，上面的代碼將創(chuàng)建一個讀取流，將其傳輸?shù)絚rypto createDecipher（）流中（使用相同的秘密），將其輸出管道輸入到zlib createGunzip（）流中， 然后將文件寫回到?jīng)]有擴(kuò)展名的文件中。

以上就是全部了，謝謝閱讀！！

翻譯自Node.js Streams: Everything you need to know

創(chuàng)建了一個程序員交流微信群，大家進(jìn)群交流IT技術(shù)

如果已過期，可以添加博主微信號15706211347，拉你進(jìn)群