把一個東西從 A 搬到 B 該怎么搬呢?
抬起來，移動到目的地，放下不就行了么。
那如果這個東西有一噸重呢?
那就一部分一部分的搬。
其實 IO 也就是搬東西，包括網絡的 IO、文件的 IO，如果數據量少，那么直接傳送全部內容就行了，但如果內容特別多，一次性加載到內存會崩潰，而且速度也慢，這時候就可以一部分一部分的處理，這就是流的思想。
各種語言基本都實現了 stream 的 api，Node.js 也是，stream api 是比較常用的，下面我們就來探究一下 stream。
本文會回答以下問題：
- Node.js 的 4 種 stream 是什么
- 生成器如何與 Readable Stream 結合
- stream 的暫停和流動
- 什么是背壓問題，如何解決
Node.js 的 4種 stream
流的直觀感受
從一個地方流到另一個地方，顯然有流出的一方和流入的一方，流出的一方就是可讀流(readable)，而流入的一方就是可寫流(writable)。
當然，也有的流既可以流入又可以流出，這種叫做雙工流(duplex)
既然可以流入又可以流出，那么是不是可以對流入的內容做下轉換再流出呢，這種流叫做轉換流(transform)
duplex 流的流入和流出內容不需要相關，而 transform 流的流入和流出是相關的，這是兩者的區別。
流的 api
Node.js 提供的 stream 就是上面介紹的那 4 種：
- conststream=require('stream');
- //可讀流
- constReadable=stream.Readable;
- //可寫流
- constWritable=stream.Writable;
- //雙工流
- constDuplex=stream.Duplex;
- //轉換流
- constTransform=stream.Transform;
它們都有要實現的方法：
- Readable 需要實現 _read 方法來返回內容
- Writable 需要實現 _write 方法來接受內容
- Duplex 需要實現 _read 和 _write 方法來接受和返回內容
- Transform 需要實現 _transform 方法來把接受的內容轉換之后返回
我們分別來看一下：
Readable
Readable 要實現 _read 方法，通過 push 返回具體的數據。
- constStream=require('stream');
- constreadableStream=Stream.Readable();
- readableStream._read=function(){
- this.push('阿門阿前一棵葡萄樹，');
- this.push('阿東阿東綠的剛發芽，');
- this.push('阿東背著那重重的的殼呀，');
- this.push('一步一步地往上爬。')
- this.push(null);
- }
- readableStream.on('data',(data)=@gt;{
- console.log(data.toString())
- });
- readableStream.on('end',()=@gt;{
- console.log('done~');
- });
當 push 一個 null 時，就代表結束流。
執行效果如下：
創建 Readable 也可以通過繼承的方式：
- constStream=require('stream');
- classReadableDongextendsStream.Readable{
- constructor(){
- super();
- }
- _read(){
- this.push('阿門阿前一棵葡萄樹，');
- this.push('阿東阿東綠的剛發芽，');
- this.push('阿東背著那重重的的殼呀，');
- this.push('一步一步地往上爬。')
- this.push(null);
- }
- }
- constreadableStream=newReadableDong();
- readableStream.on('data',(data)=@gt;{
- console.log(data.toString())
- });
- readableStream.on('end',()=@gt;{
- console.log('done~');
- });
可讀流是生成內容的，那么很自然可以和生成器結合：
- constStream=require('stream');
- classReadableDongextendsStream.Readable{
- constructor(iterator){
- super();
- this.iterator=iterator;
- }
- _read(){
- constnext=this.iterator.next();
- if(next.done){
- returnthis.push(null);
- }else{
- this.push(next.value)
- }
- }
- }
- function*songGenerator(){
- yield'阿門阿前一棵葡萄樹，';
- yield'阿東阿東綠的剛發芽，';
- yield'阿東背著那重重的的殼呀，';
- yield'一步一步地往上爬。';
- }
- constsongIterator=songGenerator();
- constreadableStream=newReadableDong(songIterator);
- readableStream.on('data',(data)=@gt;{
- console.log(data.toString())
- });
- readableStream.on('end',()=@gt;{
- console.log('done~');
- });
這就是可讀流，通過實現 _read 方法來返回內容。
Writable
Writable 要實現 _write 方法，接收寫入的內容。
- constStream=require('stream');
- constwritableStream=Stream.Writable();
- writableStream._write=function(data,enc,next){
- console.log(data.toString());
- //每秒寫一次
- setTimeout(()=@gt;{
- next();
- },1000);
- }
- writableStream.on('finish',()=@gt;console.log('done~'));
- writableStream.write('阿門阿前一棵葡萄樹，');
- writableStream.write('阿東阿東綠的剛發芽，');
- writableStream.write('阿東背著那重重的的殼呀，');
- writableStream.write('一步一步地往上爬。');
- writableStream.end();
接收寫入的內容，打印出來，并且調用 next 來處理下一個寫入的內容，這里調用 next 是異步的，可以控制頻率。
跑了一下，確實可以正常的處理寫入的內容：
這就是可寫流，通過實現 _write 方法來處理寫入的內容。
Duplex
Duplex 是可讀可寫，同時實現 _read 和 _write 就可以了
- constStream=require('stream');
- varduplexStream=Stream.Duplex();
- duplexStream._read=function(){
- this.push('阿門阿前一棵葡萄樹，');
- this.push('阿東阿東綠的剛發芽，');
- this.push('阿東背著那重重的的殼呀，');
- this.push('一步一步地往上爬。')
- this.push(null);
- }
- duplexStream._write=function(data,enc,next){
- console.log(data.toString());
- next();
- }
- duplexStream.on('data',data=@gt;console.log(data.toString()));
- duplexStream.on('end',data=@gt;console.log('readdone~'));
- duplexStream.write('阿門阿前一棵葡萄樹，');
- duplexStream.write('阿東阿東綠的剛發芽，');
- duplexStream.write('阿東背著那重重的的殼呀，');
- duplexStream.write('一步一步地往上爬。');
- duplexStream.end();
- duplexStream.on('finish',data=@gt;console.log('writedone~'));
整合了 Readable 流和 Writable 流的功能，這就是雙工流 Duplex。
Transform
Duplex 流雖然可讀可寫，但是兩者之間沒啥關聯，而有的時候需要對流入的內容做轉換之后流出，這時候就需要轉換流 Transform。
Transform 流要實現 _transform 的 api，我們實現下對內容做反轉的轉換流：
- constStream=require('stream');
- classTransformReverseextendsStream.Transform{
- constructor(){
- super()
- }
- _transform(buf,enc,next){
- constres=buf.toString().split('').reverse().join('');
- this.push(res)
- next()
- }
- }
- vartransformStream=newTransformReverse();
- transformStream.on('data',data=@gt;console.log(data.toString()))
- transformStream.on('end',data=@gt;console.log('readdone~'));
- transformStream.write('阿門阿前一棵葡萄樹');
- transformStream.write('阿東阿東綠的剛發芽');
- transformStream.write('阿東背著那重重的的殼呀');
- transformStream.write('一步一步地往上爬');
- transformStream.end()
- transformStream.on('finish',data=@gt;console.log('writedone~'));
跑了一下，效果如下：
流的暫停和流動
我們從 Readable 流中獲取內容，然后流入 Writable 流，兩邊分別做 _read 和 _write 的實現，就實現了流動。
背壓
但是 read 和 write 都是異步的，如果兩者速率不一致呢?
如果 Readable 讀入數據的速率大于 Writable 寫入速度的速率，這樣就會積累一些數據在緩沖區，如果緩沖的數據過多，就會爆掉，會丟失數據。
而如果 Readable 讀入數據的速率小于 Writable 寫入速度的速率呢?那沒關系，最多就是中間有段空閑時期。
這種讀入速率大于寫入速率的現象叫做“背壓”，或者“負壓”。也很好理解，寫入段壓力比較大，寫不進去了，會爆緩沖區，導致數據丟失。
這個緩沖區大小可以通過 readableHighWaterMark 和 writableHightWaterMark 來查看，是 16k。
解決背壓
怎么解決這種讀寫速率不一致的問題呢?
當沒寫完的時候，暫停讀就行了。這樣就不會讀入的數據越來越多，駐留在緩沖區。
readable stream 有個 readableFlowing 的屬性，代表是否自動讀入數據，默認為 true，也就是自動讀入數據，然后監聽 data 事件就可以拿到了。
當 readableFlowing 設置為 false 就不會自動讀了，需要手動通過 read 來讀入。
- readableStream.readableFlowing=false;
- letdata;
- while((data=readableStream.read())!=null){
- console.log(data.toString());
- }
但自己手動 read 比較麻煩，我們依然可以用自動流入的方式，調用 pause 和 resume 來暫停和恢復就行了。
當調用 writable stream 的 write 方法的時候會返回一個 boolean 值代表是寫入了目標還是放在了緩沖區：
- true: 數據已經寫入目標
- false：目標不可寫入，暫時放在緩沖區
我們可以判斷返回 false 的時候就 pause，然后等緩沖區清空了就 resume：
- constrs=fs.createReadStream(src);
- constws=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();
- });
這樣就能達到根據寫入速率暫停和恢復讀入速率的功能，解決了背壓問題。
pipe 有背壓問題么?
平時我們經常會用 pipe 來直接把 Readable 流對接到 Writable 流，但是好像也沒遇到過背壓問題，其實是 pipe 內部已經做了讀入速率的動態調節了。
- constrs=fs.createReadStream(src);
- constws=fs.createWriteStream(dst);
- rs.pipe(ws);
總結
流是傳輸數據時常見的思想，就是一部分一部分的傳輸內容，是文件讀寫、網絡通信的基礎概念。
Node.js 也提供了 stream 的 api，包括 Readable 可讀流、Writable 可寫流、Duplex 雙工流、Transform 轉換流。它們分別實現 _read、_write、_read + _write、_transform 方法，來做數據的返回和處理。
創建 Readable 對象既可以直接調用 Readable api 創建，然后重寫 _read 方法，也可以繼承 Readable 實現一個子類，之后實例化。其他流同理。(Readable 可以很容易的和 generator 結合)
當讀入的速率大于寫入速率的時候就會出現“背壓”現象，會爆緩沖區導致數據丟失，解決的方式是根據 write 的速率來動態 pause 和 resume 可讀流的速率。pipe 就沒有這個問題，因為內部做了處理。
流是掌握 IO 繞不過去的一個概念，而背壓問題也是流很常見的問題，遇到了數據丟失可以考慮是否發生了背壓。希望這篇文章能夠幫大家理清思路，真正掌握 stream!
原文鏈接：https://mp.weixin.qq.com/s/9bf53AR2ephzcMTv5P5dJg
