Node.js 中流操作實(shí)踐

chaos_G 發(fā)布于2019-08-23 11:52 / 3018人閱讀

摘要：事件的觸發(fā)頻次同樣是由實(shí)現(xiàn)者決定，譬如在進(jìn)行文件讀取時(shí)，可能每行都會(huì)觸發(fā)一次而在請(qǐng)求處理時(shí)，可能數(shù)的數(shù)據(jù)才會(huì)觸發(fā)一次。如果有參數(shù)傳入，它會(huì)讓可讀流停止流向某個(gè)特定的目的地，否則，它會(huì)移除所有目的地。

本文節(jié)選自 Node.js CheatSheet | Node.js 語(yǔ)法基礎(chǔ)、框架使用與實(shí)踐技巧，也可以閱讀 JavaScript CheatSheet 或者 現(xiàn)代 Web 開(kāi)發(fā)基礎(chǔ)與工程實(shí)踐 了解更多 JavaScript/Node.js 的實(shí)際應(yīng)用。

Stream 是 Node.js 中的基礎(chǔ)概念，類似于 EventEmitter，專注于 IO 管道中事件驅(qū)動(dòng)的數(shù)據(jù)處理方式；類比于數(shù)組或者映射，Stream 也是數(shù)據(jù)的集合，只不過(guò)其代表了不一定正在內(nèi)存中的數(shù)據(jù)。。Node.js 的 Stream 分為以下類型：

Readable Stream: 可讀流，數(shù)據(jù)的產(chǎn)生者，譬如 process.stdin

Writable Stream: 可寫流，數(shù)據(jù)的消費(fèi)者，譬如 process.stdout 或者 process.stderr

Duplex Stream: 雙向流，即可讀也可寫

Transform Stream: 轉(zhuǎn)化流，數(shù)據(jù)的轉(zhuǎn)化者

Stream 本身提供了一套接口規(guī)范，很多 Node.js 中的內(nèi)建模塊都遵循了該規(guī)范，譬如著名的 fs 模塊，即是使用 Stream 接口來(lái)進(jìn)行文件讀寫；同樣的，每個(gè) HTTP 請(qǐng)求是可讀流，而 HTTP 響應(yīng)則是可寫流。

Readable Stream

const stream = require("stream");
const fs = require("fs");

const readableStream = fs.createReadStream(process.argv[2], {
  encoding: "utf8"
});

// 手動(dòng)設(shè)置流數(shù)據(jù)編碼
// readableStream.setEncoding("utf8");

let wordCount = 0;

readableStream.on("data", function(data) {
  wordCount += data.split(/s{1,}/).length;
});

readableStream.on("end", function() {
  // Don"t count the end of the file.
  console.log("%d %s", --wordCount, process.argv[2]);
});

當(dāng)我們創(chuàng)建某個(gè)可讀流時(shí)，其還并未開(kāi)始進(jìn)行數(shù)據(jù)流動(dòng)；添加了 data 的事件監(jiān)聽(tīng)器，它才會(huì)變成流動(dòng)態(tài)的。在這之后，它就會(huì)讀取一小塊數(shù)據(jù)，然后傳到我們的回調(diào)函數(shù)里面。 data 事件的觸發(fā)頻次同樣是由實(shí)現(xiàn)者決定，譬如在進(jìn)行文件讀取時(shí)，可能每行都會(huì)觸發(fā)一次；而在 HTTP 請(qǐng)求處理時(shí)，可能數(shù) KB 的數(shù)據(jù)才會(huì)觸發(fā)一次。可以參考 nodejs/readable-stream/_stream_readable 中的相關(guān)實(shí)現(xiàn)，發(fā)現(xiàn) on 函數(shù)會(huì)觸發(fā) resume 方法，該方法又會(huì)調(diào)用 flow 函數(shù)進(jìn)行流讀取：

// function on
if (ev === "data") {
  // Start flowing on next tick if stream isn"t explicitly paused
  if (this._readableState.flowing !== false) this.resume();
}
...
// function flow
while (state.flowing && stream.read() !== null) {}

我們還可以監(jiān)聽(tīng) readable 事件，然后手動(dòng)地進(jìn)行數(shù)據(jù)讀取：

let data = "";
let chunk;
readableStream.on("readable", function() {
  while ((chunk = readableStream.read()) != null) {
    data += chunk;
  }
});
readableStream.on("end", function() {
  console.log(data);
});

Readable Stream 還包括如下常用的方法：

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ì)移除所有目的地。

在日常開(kāi)發(fā)中，我們可以用 stream-wormhole 來(lái)模擬消耗可讀流：

sendToWormhole(readStream, true);

Writable Stream

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

writableStream.end();

當(dāng) end() 被調(diào)用時(shí)，所有數(shù)據(jù)會(huì)被寫入，然后流會(huì)觸發(fā)一個(gè) finish 事件。注意在調(diào)用 end() 之后，你就不能再往可寫流中寫入數(shù)據(jù)了。

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

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

process.stdin.pipe(outStream);

Writable Stream 中同樣包含一些與 Readable Stream 相關(guān)的重要事件：

error: 在寫入或鏈接發(fā)生錯(cuò)誤時(shí)觸發(fā)

pipe: 當(dāng)可讀流鏈接到可寫流時(shí)，這個(gè)事件會(huì)觸發(fā)

unpipe: 在可讀流調(diào)用 unpipe 時(shí)會(huì)觸發(fā)

Pipe | 管道

const fs = require("fs");

const inputFile = fs.createReadStream("REALLY_BIG_FILE.x");
const outputFile = fs.createWriteStream("REALLY_BIG_FILE_DEST.x");

// 當(dāng)建立管道時(shí)，才發(fā)生了流的流動(dòng)
inputFile.pipe(outputFile);

多個(gè)管道順序調(diào)用，即是構(gòu)建了鏈接(Chaining):

const fs = require("fs");
const zlib = require("zlib");
fs.createReadStream("input.txt.gz")
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream("output.txt"));

管道也常用于 Web 服務(wù)器中的文件處理，以 Egg.js 中的應(yīng)用為例，我們可以從 Context 中獲取到文件流并將其傳入到可寫文件流中：