NodeJS中的可讀流是什么

這篇文章主要介紹了NodeJS中的可讀流是什么，具有一定借鑒價值，需要的朋友可以參考下。希望大家閱讀完這篇文章后大有收獲。下面讓小編帶著大家一起了解一下。

什么是可讀流

可讀流是生產(chǎn)數(shù)據(jù)用來供程序消費的流。我們常見的數(shù)據(jù)生產(chǎn)方式有讀取磁盤文件、讀取網(wǎng)絡請求內(nèi)容等，看一下前面介紹什么是流用的例子：

const rs = fs.createReadStream(filePath);

rs 就是一個可讀流，其生產(chǎn)數(shù)據(jù)的方式是讀取磁盤的文件，我們常見的控制臺 process.stdin 也是一個可讀流：

process.stdin.pipe(process.stdout);

通過簡單的一句話可以把控制臺的輸入打印出來，process.stdin 生產(chǎn)數(shù)據(jù)的方式是讀取用戶在控制臺的輸入。

回頭再看一下我們對可讀流的定義：可讀流是生產(chǎn)數(shù)據(jù)用來供程序消費的流。

自定義可讀流

除了系統(tǒng)提供給我們的 fs.CreateReadStream 我們還經(jīng)常使用 gulp 或者 vinyl-fs 提供的 src 方法

gulp.src(['*.js', 'dist/**/*.scss'])

如果我們想自己以某種特定的方式生產(chǎn)數(shù)據(jù)，交給程序消費，那么改如何開始呢？

簡單兩步即可

1. 繼承 sream 模塊的 Readable 類
2. 重寫 _read 方法，調(diào)用 this.push 將生產(chǎn)的數(shù)據(jù)放入待讀取隊列

Readable 類已經(jīng)把可讀流要做的大部分工作完成，我們只需要繼承它，然后把生產(chǎn)數(shù)據(jù)的方式寫在 _read 方法里就可以實現(xiàn)一個自定義的可讀流。

如果我們想實現(xiàn)一個每 100 毫秒生產(chǎn)一個隨機數(shù)的流（沒什么用處）

const Readable = require('stream').Readable;

class RandomNumberStream extends Readable {
    constructor(max) {
        super()
    }

    _read() {
        const ctx = this;
        setTimeout(() => {
            const randomNumber = parseInt(Math.random() * 10000);

            // 只能 push 字符串或 Buffer，為了方便顯示打一個回車
            ctx.push(`${randomNumber}\n`);
        }, 100);
    }
}

module.exports = RandomNumberStream;

類繼承部分代碼很簡單，主要看一下 _read 方法的實現(xiàn)，有幾個值得注意的地方

1. Readable 類中默認有 _read 方法的實現(xiàn)，不過什么都沒有做，我們做的是覆蓋重寫
2. _read 方法有一個參數(shù) size，用來向 read 方法指定應該讀取多少數(shù)據(jù)返回，不過只是一個參考數(shù)據(jù)，很多實現(xiàn)忽略此參數(shù)，我們這里也忽略了，后面會詳細提到
3. 通過 this.push 向緩沖區(qū)推送數(shù)據(jù)，緩沖區(qū)概念后面會提到，暫時理解為擠到了水管中可消費了
4. push 的內(nèi)容只能是字符串或者 Buffer，不能是數(shù)字
5. push 方法有第二個參數(shù) encoding，用于第一個參數(shù)是字符串時指定 encoding

執(zhí)行一下看看效果

const RandomNumberStream = require('./RandomNumberStream');

const rns = new RandomNumberStream();

rns.pipe(process.stdout);

這樣可以看到數(shù)字源源不斷的顯示到了控制臺上，我們實現(xiàn)了一個產(chǎn)生隨機數(shù)的可讀流，還有幾個小問題待解決

如何停下來

我們每隔 100 毫秒向緩沖區(qū)推送一個數(shù)字，那么就像讀取一個本地文件總有讀完的時候，如何停下來標識數(shù)據(jù)讀取完畢？

向緩沖區(qū) push 一個 null 就可以。我們修改一下代碼，允許消費者定義需要多少個隨機數(shù)字：

const Readable = require('stream').Readable;

class RandomNumberStream extends Readable {
    constructor(max) {
        super()
        this.max = max;
    }

    _read() {
        const ctx = this;

        setTimeout(() => {
            if (ctx.max) {
                const randomNumber = parseInt(Math.random() * 10000);

                // 只能 push 字符串或 Buffer，為了方便顯示打一個回車
                ctx.push(`${randomNumber}\n`);
                ctx.max -= 1;
            } else {
                ctx.push(null);
            }
        }, 100);
    }
}

module.exports = RandomNumberStream;

我們使用了一個 max 的標識，允許消費者指定需要的字符數(shù)，在實例化的時候指定即可

const RandomNumberStream = require('./RandomNumberStream');

const rns = new RandomNumberStream(5);

rns.pipe(process.stdout);

這樣可以看到控制臺只打印了 5 個字符

為什么是 setTimeout 而不是 setInterval

細心的同學可能注意到，我們每隔 100 毫秒生產(chǎn)一個隨機數(shù)并不是調(diào)用的 setInterval，而是使用的 setTimeout，為什么僅僅是延時了一下并沒有重復生產(chǎn)，結(jié)果卻是正確的呢？

這就需要了解流的兩種工作方式

1. 流動模式：數(shù)據(jù)由底層系統(tǒng)讀出，并盡可能快地提供給應用程序
2. 暫停模式：必須顯示地調(diào)用 read() 方法來讀取若干數(shù)據(jù)塊

流在默認狀態(tài)下是處于暫停模式的，也就是需要程序顯式的調(diào)用 read() 方法，可我們的例子中并沒有調(diào)用就可以得到數(shù)據(jù)，因為我們的流通過 pipe() 方法切換成了流動模式，這樣我們的 _read() 方法會自動被反復調(diào)用，直到數(shù)據(jù)讀取完畢，所以我們每次 _read() 方法里面只需要讀取一次數(shù)據(jù)即可。

流動模式和暫停模式切換

流從默認的暫停模式切換到流動模式可以使用以下幾種方式：

1. 通過添加 data 事件監(jiān)聽器來啟動數(shù)據(jù)監(jiān)聽
2. 調(diào)用 resume() 方法啟動數(shù)據(jù)流
3. 調(diào)用 pipe() 方法將數(shù)據(jù)轉(zhuǎn)接到另一個 可寫流

從流動模式切換為暫停模式又兩種方法：

1. 在流沒有 pipe() 時，調(diào)用 pause() 方法可以將流暫停
2. pipe() 時，需要移除所有 data 事件的監(jiān)聽，再調(diào)用 unpipe() 方法

data 事件

使用了 pipe() 方法后數(shù)據(jù)就從可讀流進入了可寫流，但對我們好像是個黑盒，數(shù)據(jù)究竟是怎么流向的呢？我們看到切換流動模式和暫停模式的時候有兩個重要的名詞

1. 流動模式對應的 data 事件
2. 暫停模式對應的 read() 方法

這兩個機制是我們能夠驅(qū)動數(shù)據(jù)流動的原因，先來看一下流動模式 data 事件，一旦我們監(jiān)聽了可讀流的 data 時、事件，流就進入了流動模式，我們可以改寫一下上面調(diào)用流的代碼

const RandomNumberStream = require('./RandomNumberStream');

const rns = new RandomNumberStream(5);

rns.on('data', chunk => {
  console.log(chunk);
});

這樣我們可以看到控制臺打印出了類似下面的結(jié)果

<Buffer 39 35 37 0a>
<Buffer 31 30 35 37 0a>
<Buffer 38 35 31 30 0a>
<Buffer 33 30 35 35 0a>
<Buffer 34 36 34 32 0a>

當可讀流生產(chǎn)出可供消費的數(shù)據(jù)后就會觸發(fā) data 事件，data 事件監(jiān)聽器綁定后，數(shù)據(jù)會被盡可能地傳遞。data 事件的監(jiān)聽器可以在第一個參數(shù)收到可讀流傳遞過來的 Buffer 數(shù)據(jù)，這也就是我們打印的 chunk，如果想顯示為數(shù)字，可以調(diào)用 Buffer 的 toString() 方法。

當數(shù)據(jù)處理完成后還會觸發(fā)一個 end 事件，應為流的處理不是同步調(diào)用，所以如果我們希望完事后做一些事情就需要監(jiān)聽這個事件，我們在代碼最后追加一句：

rns.on('end', () => {
  console.log('done');
});

這樣可以在數(shù)據(jù)接收完了顯示 'done'

當然數(shù)據(jù)處理過程中出現(xiàn)了錯誤會觸發(fā) error 事件，我們同樣可以監(jiān)聽，做異常處理：

rns.on('error', (err) => {
  console.log(err);
});

read(size)

流在暫停模式下需要程序顯式調(diào)用 read() 方法才能得到數(shù)據(jù)。read() 方法會從內(nèi)部緩沖區(qū)中拉取并返回若干數(shù)據(jù)，當沒有更多可用數(shù)據(jù)時，會返回null。

使用 read() 方法讀取數(shù)據(jù)時，如果傳入了 size 參數(shù)，那么它會返回指定字節(jié)的數(shù)據(jù)；當指定的size字節(jié)不可用時，則返回null。如果沒有指定size參數(shù)，那么會返回內(nèi)部緩沖區(qū)中的所有數(shù)據(jù)。

現(xiàn)在有一個矛盾了，在流動模式下流生產(chǎn)出了數(shù)據(jù)，然后觸發(fā) data 事件通知給程序，這樣很方便。在暫停模式下需要程序去讀取，那么就有一種可能是讀取的時候還沒生產(chǎn)好，如果我們才用輪詢的方式未免效率有些低。

NodeJS 為我們提供了一個 readable 的事件，事件在可讀流準備好數(shù)據(jù)的時候觸發(fā)，也就是先監(jiān)聽這個事件，收到通知又數(shù)據(jù)了我們再去讀取就好了：

const rns = new RandomNumberStream(5);

rns.on('readable', () => {
  let chunk;
  while((chunk = rns.read()) !== null){
    console.log(chunk);
  }
});

這樣我們同樣可以讀取到數(shù)據(jù)，值得注意的一點是并不是每次調(diào)用 read() 方法都可以返回數(shù)據(jù)，前面提到了如果可用的數(shù)據(jù)沒有達到 size 那么返回 null，所以我們在程序中加了個判斷。

數(shù)據(jù)會不會漏掉

開始使用流動模式的時候我經(jīng)常會擔心一個問題，上面代碼中可讀流在創(chuàng)建好的時候就生產(chǎn)數(shù)據(jù)了，那么會不會在我們綁定 readable 事件之前就生產(chǎn)了某些數(shù)據(jù)，觸發(fā)了 readable 事件，我們還沒有綁定，這樣不是極端情況下會造成開頭數(shù)據(jù)的丟失嘛

可事實并不會，按照 NodeJS event loop 我們創(chuàng)建流和調(diào)用事件監(jiān)聽在一個事件隊列里面，兒生產(chǎn)數(shù)據(jù)由于涉及到異步操作，已經(jīng)處于了下一個事件隊列，我們監(jiān)聽事件再慢也會比數(shù)據(jù)生產(chǎn)塊，數(shù)據(jù)不會丟失。

看到這里，大家其實對 data事件、readable事件觸發(fā)時機， read() 方法每次讀多少數(shù)據(jù)，什么時候返回 null 還有又一定的疑問，因為到現(xiàn)在為止我們接觸到的仍然是一個黑盒，后面我們介紹了可寫流后會在 back pressure 機制部分對這些內(nèi)部細節(jié)結(jié)合源碼詳細講解，且聽下回分解吧。

感謝你能夠認真閱讀完這篇文章，希望小編分享NodeJS中的可讀流是什么內(nèi)容對大家有幫助，同時也希望大家多多支持億速云，關注億速云行業(yè)資訊頻道，遇到問題就找億速云，詳細的解決方法等著你來學習!