本篇文章帶大家了解一下Node.js中的4種 stream，看看怎么解決爆緩沖區(qū)的“背壓”問題，有需要的朋友可以去學(xué)習(xí)了解一下~

把一個(gè)東西從 A 搬到 B 該怎么搬呢？

抬起來，移動到目的地，放下不就行了么。

那如果這個(gè)東西有一噸重呢？

那就一部分一部分的搬。

其實(shí) IO 也就是搬東西，包括網(wǎng)絡(luò)的 IO、文件的 IO，如果數(shù)據(jù)量少，那么直接傳送全部內(nèi)容就行了，但如果內(nèi)容特別多，一次性加載到內(nèi)存會崩潰，而且速度也慢，這時(shí)候就可以一部分一部分的處理，這就是流的思想?！就扑]學(xué)習(xí)：《nodejs 教程》】

各種語言基本都實(shí)現(xiàn)了 stream 的 api，Node.js 也是，stream api 是比較常用的，下面我們就來探究一下 stream。

本文會回答以下問題：

• Node.js 的 4 種 stream 是什么
• 生成器如何與 Readable Stream 結(jié)合
• stream 的暫停和流動
• 什么是背壓問題，如何解決

Node.js 的 4種 stream

流的直觀感受

從一個(gè)地方流到另一個(gè)地方，顯然有流出的一方和流入的一方，流出的一方就是可讀流(readable)，而流入的一方就是可寫流(writable)。

當(dāng)然，也有的流既可以流入又可以流出，這種叫做雙工流（duplex）

既然可以流入又可以流出，那么是不是可以對流入的內(nèi)容做下轉(zhuǎn)換再流出呢，這種流叫做轉(zhuǎn)換流（transform）

duplex 流的流入和流出內(nèi)容不需要相關(guān)，而 transform 流的流入和流出是相關(guān)的，這是兩者的區(qū)別。

流的 api

Node.js 提供的 stream 就是上面介紹的那 4 種：

const stream = require('stream');  // 可讀流 const Readable = stream.Readable; // 可寫流 const Writable = stream.Writable; // 雙工流 const Duplex = stream.Duplex; // 轉(zhuǎn)換流 const Transform = stream.Transform;

它們都有要實(shí)現(xiàn)的方法：

• Readable 需要實(shí)現(xiàn) _read 方法來返回內(nèi)容
• Writable 需要實(shí)現(xiàn) _write 方法來接受內(nèi)容
• Duplex 需要實(shí)現(xiàn) _read 和 _write 方法來接受和返回內(nèi)容
• Transform 需要實(shí)現(xiàn) _transform 方法來把接受的內(nèi)容轉(zhuǎn)換之后返回

我們分別來看一下：

Readable

Readable 要實(shí)現(xiàn) _read 方法，通過 push 返回具體的數(shù)據(jù)。

const Stream = require('stream');  const readableStream = Stream.Readable();  readableStream._read = function() {     this.push('阿門阿前一棵葡萄樹，');     this.push('阿東阿東綠的剛發(fā)芽，');     this.push('阿東背著那重重的的殼呀，');     this.push('一步一步地往上爬。')     this.push(null); }  readableStream.on('data', (data)=> {     console.log(data.toString()) });  readableStream.on('end', () => {     console.log('done~'); });

當(dāng) push 一個(gè) null 時(shí)，就代表結(jié)束流。

執(zhí)行效果如下：

創(chuàng)建 Readable 也可以通過繼承的方式：

const Stream = require('stream');  class ReadableDong extends Stream.Readable {      constructor() {         super();     }      _read() {         this.push('阿門阿前一棵葡萄樹，');         this.push('阿東阿東綠的剛發(fā)芽，');         this.push('阿東背著那重重的的殼呀，');         this.push('一步一步地往上爬。')         this.push(null);     }  }  const readableStream = new ReadableDong();  readableStream.on('data', (data)=> {     console.log(data.toString()) });  readableStream.on('end', () => {     console.log('done~'); });

可讀流是生成內(nèi)容的，那么很自然可以和生成器結(jié)合：

const Stream = require('stream');  class ReadableDong extends Stream.Readable {      constructor(iterator) {         super();         this.iterator = iterator;     }      _read() {         const next = this.iterator.next();         if(next.done) {             return this.push(null);         } else {             this.push(next.value)         }     }  }  function *songGenerator() {     yield '阿門阿前一棵葡萄樹，';     yield '阿東阿東綠的剛發(fā)芽，';     yield '阿東背著那重重的的殼呀，';     yield '一步一步地往上爬。'; }  const songIterator = songGenerator();  const readableStream = new ReadableDong(songIterator);  readableStream.on('data', (data)=> {     console.log(data.toString()) });  readableStream.on('end', () => {     console.log('done~'); });

這就是可讀流，通過實(shí)現(xiàn) _read 方法來返回內(nèi)容。

Writable

Writable 要實(shí)現(xiàn) _write 方法，接收寫入的內(nèi)容。

const Stream = require('stream');  const writableStream = Stream.Writable();  writableStream._write = function (data, enc, next) {    console.log(data.toString());    // 每秒寫一次    setTimeout(() => {        next();    }, 1000); }  writableStream.on('finish', () => console.log('done~'));  writableStream.write('阿門阿前一棵葡萄樹，'); writableStream.write('阿東阿東綠的剛發(fā)芽，'); writableStream.write('阿東背著那重重的的殼呀，'); writableStream.write('一步一步地往上爬。'); writableStream.end();

接收寫入的內(nèi)容，打印出來，并且調(diào)用 next 來處理下一個(gè)寫入的內(nèi)容，這里調(diào)用 next 是異步的，可以控制頻率。

跑了一下，確實(shí)可以正常的處理寫入的內(nèi)容：

這就是可寫流，通過實(shí)現(xiàn) _write 方法來處理寫入的內(nèi)容。

Duplex

Duplex 是可讀可寫，同時(shí)實(shí)現(xiàn) _read 和 _write 就可以了

const Stream = require('stream');  var duplexStream = Stream.Duplex();  duplexStream._read = function () {     this.push('阿門阿前一棵葡萄樹，');     this.push('阿東阿東綠的剛發(fā)芽，');     this.push('阿東背著那重重的的殼呀，');     this.push('一步一步地往上爬。')     this.push(null); }  duplexStream._write = function (data, enc, next) {     console.log(data.toString());     next(); }  duplexStream.on('data', data => console.log(data.toString())); duplexStream.on('end', data => console.log('read done~'));  duplexStream.write('阿門阿前一棵葡萄樹，'); duplexStream.write('阿東阿東綠的剛發(fā)芽，'); duplexStream.write('阿東背著那重重的的殼呀，'); duplexStream.write('一步一步地往上爬。'); duplexStream.end();  duplexStream.on('finish', data => console.log('write done~'));

整合了 Readable 流和 Writable 流的功能，這就是雙工流 Duplex。

Transform

Duplex 流雖然可讀可寫，但是兩者之間沒啥關(guān)聯(lián)，而有的時(shí)候需要對流入的內(nèi)容做轉(zhuǎn)換之后流出，這時(shí)候就需要轉(zhuǎn)換流 Transform。

Transform 流要實(shí)現(xiàn) _transform 的 api，我們實(shí)現(xiàn)下對內(nèi)容做反轉(zhuǎn)的轉(zhuǎn)換流：

const Stream = require('stream');  class TransformReverse extends Stream.Transform {    constructor() {     super()   }    _transform(buf, enc, next) {     const res = buf.toString().split('').reverse().join('');     this.push(res)     next()   } }  var transformStream = new TransformReverse();  transformStream.on('data', data => console.log(data.toString())) transformStream.on('end', data => console.log('read done~'));  transformStream.write('阿門阿前一棵葡萄樹'); transformStream.write('阿東阿東綠的剛發(fā)芽'); transformStream.write('阿東背著那重重的的殼呀'); transformStream.write('一步一步地往上爬'); transformStream.end()  transformStream.on('finish', data => console.log('write done~'));

跑了一下，效果如下：

流的暫停和流動

我們從 Readable 流中獲取內(nèi)容，然后流入 Writable 流，兩邊分別做 _read 和 _write 的實(shí)現(xiàn)，就實(shí)現(xiàn)了流動。

背壓

但是 read 和 write 都是異步的，如果兩者速率不一致呢？

如果 Readable 讀入數(shù)據(jù)的速率大于 Writable 寫入速度的速率，這樣就會積累一些數(shù)據(jù)在緩沖區(qū)，如果緩沖的數(shù)據(jù)過多，就會爆掉，會丟失數(shù)據(jù)。

而如果 Readable 讀入數(shù)據(jù)的速率小于 Writable 寫入速度的速率呢？那沒關(guān)系，最多就是中間有段空閑時(shí)期。

這種讀入速率大于寫入速率的現(xiàn)象叫做“背壓”，或者“負(fù)壓”。也很好理解，寫入段壓力比較大，寫不進(jìn)去了，會爆緩沖區(qū)，導(dǎo)致數(shù)據(jù)丟失。

這個(gè)緩沖區(qū)大小可以通過 readableHighWaterMark 和 writableHightWaterMark 來查看，是 16k。

解決背壓

怎么解決這種讀寫速率不一致的問題呢？

當(dāng)沒寫完的時(shí)候，暫停讀就行了。這樣就不會讀入的數(shù)據(jù)越來越多，駐留在緩沖區(qū)。

readable stream 有個(gè) readableFlowing 的屬性，代表是否自動讀入數(shù)據(jù)，默認(rèn)為 true，也就是自動讀入數(shù)據(jù)，然后監(jiān)聽 data 事件就可以拿到了。

當(dāng) readableFlowing 設(shè)置為 false 就不會自動讀了，需要手動通過 read 來讀入。

readableStream.readableFlowing = false;  let data; while((data = readableStream.read()) != null) {     console.log(data.toString()); }

但自己手動 read 比較麻煩，我們依然可以用自動流入的方式，調(diào)用 pause 和 resume 來暫停和恢復(fù)就行了。

當(dāng)調(diào)用 writable stream 的 write 方法的時(shí)候會返回一個(gè) boolean 值代表是寫入了目標(biāo)還是放在了緩沖區(qū)：

• true: 數(shù)據(jù)已經(jīng)寫入目標(biāo)
• false：目標(biāo)不可寫入，暫時(shí)放在緩沖區(qū)

我們可以判斷返回 false 的時(shí)候就 pause，然后等緩沖區(qū)清空了就 resume：

const rs = fs.createReadStream(src); const ws = fs.createWriteStream(dst);  rs.on('data', function (chunk) {     if (ws.write(chunk) === false) {         rs.pause();     } });  rs.on('end', function () {     ws.end(); });  ws.on('drain', function () {     rs.resume(); });

這樣就能達(dá)到根據(jù)寫入速率暫停和恢復(fù)讀入速率的功能，解決了背壓問題。

pipe 有背壓問題么？

平時(shí)我們經(jīng)常會用 pipe 來直接把 Readable 流對接到 Writable 流，但是好像也沒遇到過背壓問題，其實(shí)是 pipe 內(nèi)部已經(jīng)做了讀入速率的動態(tài)調(diào)節(jié)了。

const rs = fs.createReadStream(src); const ws = fs.createWriteStream(dst);  rs.pipe(ws);

總結(jié)

流是傳輸數(shù)據(jù)時(shí)常見的思想，就是一部分一部分的傳輸內(nèi)容，是文件讀寫、網(wǎng)絡(luò)通信的基礎(chǔ)概念。

Node.js 也提供了 stream 的 api，包括 Readable 可讀流、Writable 可寫流、Duplex 雙工流、Transform 轉(zhuǎn)換流。它們分別實(shí)現(xiàn) _read、_write、_read + _write、_transform 方法，來做數(shù)據(jù)的返回和處理。

創(chuàng)建 Readable 對象既可以直接調(diào)用 Readable api 創(chuàng)建，然后重寫 _read 方法，也可以繼承 Readable 實(shí)現(xiàn)一個(gè)子類，之后實(shí)例化。其他流同理。（Readable 可以很容易的和 generator 結(jié)合）

當(dāng)讀入的速率大于寫入速率的時(shí)候就會出現(xiàn)“背壓”現(xiàn)象，會爆緩沖區(qū)導(dǎo)致數(shù)據(jù)丟失，解決的方式是根據(jù) write 的速率來動態(tài) pause 和 resume 可讀流的速率。pipe 就沒有這個(gè)問題，因?yàn)閮?nèi)部做了處理。

流是掌握 IO 繞不過去的一個(gè)概念，而背壓問題也是流很常見的問題，遇到了數(shù)據(jù)丟失可以考慮是否發(fā)生了背壓。希望這篇文章能夠幫大家理清思路，真正掌握 stream！