Node中可讀流的使用方法

可讀流是生產數據用來供程序消費的流。常見的數據生產方式有讀取磁盤文件、讀取網絡請求內容等，看一下前面介紹什么是流用的例子：

const rs = fs.createReadStream(filePath);

rs 就是一個可讀流，其生產數據的方式是讀取磁盤的文件，控制臺 process.stdin 也是一個可讀流：

process.stdin.pipe(process.stdout);

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

回頭再看一下對可讀流的定義：

可讀流是生產數據用來供程序消費的流。

自定義可讀流

除了系統提供的

fs.CreateReadStream

使用過 gulp 或者 vinyl-fs 提供的 src 方法時候也在使用可讀流

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

如果希望自己以某種特定的方式生產數據，交給程序消費，那么改如何開始呢？

簡單兩步即可

1. 繼承 sream 模塊的

   Readable

   類
2. 重寫

   _read

   方法，調用

   this.push

   將生產的數據放入待讀取隊列

Readable 類已經把可讀流要做的大部分工作完成，只需要繼承它，然后把生產數據的方式寫在 _read 方法里就可以實現一個自定義的可讀流。

舉個例子：實現一個每 100 毫秒生產一個隨機數的流（沒什么用處）

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 方法的實現，有幾個值得注意的地方

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

執行一下看看效果

const RandomNumberStream = require('./RandomNumberStream');
const rns = new RandomNumberStream();
rns.pipe(process.stdout);

這樣可以看到數字源源不斷的顯示到了控制臺上，實現了一個產生隨機數的可讀流，還有幾個小問題待解決

如何停下來

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

向緩沖區 push 一個 null 就可以，修改一下代碼，允許消費者定義需要多少個隨機數字：

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 的標識，允許消費者指定需要的字符數，在實例化的時候指定即可

const RandomNumberStream = require('./');
const rns = new RandomNumberStream(5);
rns.pipe(process.stdout);

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

為什么是 setTimeout 而不是 setInterval

細心的同學可能注意到，每隔 100 毫秒生產一個隨機數并不是調用的 setInterval，而是使用的 setTimeout，為什么僅僅是延時了一下并沒有重復生產，結果卻是正確的呢？

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

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

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

流動模式和暫停模式切換

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

1. 通過添加 data 事件監聽器來啟動數據監聽
2. 調用 resume() 方法啟動數據流
3. 調用 pipe() 方法將數據轉接到另一個可寫流

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

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

data 事件

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

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

這兩個機制是程序能夠驅動數據流動的原因，先來看一下流動模式 data 事件，一旦監聽了可讀流的 data 事件，流就進入了流動模式，可以改寫一下上面調用流的代碼

const RandomNumberStream = require('./RandomNumberStream');
const rns = new RandomNumberStream(5);
rns.on('data', chunk => {
  console.log(chunk);
});

這樣可以看到控制臺打印出了類似下面的結果

<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>

當可讀流生產出可供消費的數據后就會觸發 data 事件，data 事件監聽器綁定后，數據會被盡可能地傳遞。data 事件的監聽器可以在第一個參數收到可讀流傳遞過來的 Buffer 數據，這也就是控制臺打印的 chunk，如果想顯示為數字，可以調用 Buffer 的 toString() 方法

當數據處理完成后還會觸發一個

end

事件，因為流的處理不是同步調用，所以如果希望完事后做一些事情就需要監聽這個事件，在代碼最后追加一句：

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

這樣可以在數據接收完了顯示 done ，當然數據處理過程中出現了錯誤會觸發 error 事件，可以監聽做異常處理：

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

read(size)

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

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

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

NodeJS 提供了一個

readable的事件，事件在可讀流準備好數據的時候觸發，也就是先監聽這個事件，收到通知有數據了再去讀取就好了：

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

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

數據會不會漏掉

const stream = fs.createReadStream('/dev/input/event0');
stream.on('readable', callback);復制代碼

在流動模式會不會有這樣的問題：可讀流在創建好的時候就生產數據了，如果在綁定 readable 事件之前就生產了某些數據，觸發了 readable 事件，在極端情況下會造成數據丟失嗎？

事實并不會，按照 NodeJS event loop 程序創建流和調用事件監聽在一個事件隊列里面，生產數據和事件監聽都是異步操作，而 on 監聽事件使用了 process.nextTick 會保證在數據生產之前被綁定好，相關知識可以看定時器章節中對 event loop 的解讀

到這里可能對 data事件、readable事件觸發時機， read() 方法每次讀多少數據，什么時候返回 null 還有一定的疑問，在后續可寫流章節會在 back pressure 部分結合源碼介紹相關機制

以上就是Node.js中Stream-可讀流的使用的詳細內容，更多請關注億速云其它相關文章！