這篇文章主要介紹了Node.js中有幾種stream,具有一定借鑒價(jià)值,感興趣的朋友可以參考下,希望大家閱讀完這篇文章之后大有收獲,下面讓小編帶著大家一起了解一下。
專注于為中小企業(yè)提供做網(wǎng)站、網(wǎng)站制作服務(wù),電腦端+手機(jī)端+微信端的三站合一,更高效的管理,為中小企業(yè)象山免費(fèi)做網(wǎng)站提供優(yōu)質(zhì)的服務(wù)。我們立足成都,凝聚了一批互聯(lián)網(wǎng)行業(yè)人才,有力地推動(dòng)了近1000家企業(yè)的穩(wěn)健成長,幫助中小企業(yè)通過網(wǎng)站建設(shè)實(shí)現(xiàn)規(guī)模擴(kuò)充和轉(zhuǎn)變。
從一個(gè)地方流到另一個(gè)地方,顯然有流出的一方和流入的一方,流出的一方就是可讀流(readable),而流入的一方就是可寫流(writable)。
當(dāng)然,也有的流既可以流入又可以流出,這種叫做雙工流(duplex)
既然可以流入又可以流出,那么是不是可以對(duì)流入的內(nèi)容做下轉(zhuǎn)換再流出呢,這種流叫做轉(zhuǎn)換流(transform)
duplex 流的流入和流出內(nèi)容不需要相關(guān),而 transform 流的流入和流出是相關(guān)的,這是兩者的區(qū)別。
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 要實(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 要實(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 是可讀可寫,同時(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。
Duplex 流雖然可讀可寫,但是兩者之間沒啥關(guān)聯(lián),而有的時(shí)候需要對(duì)流入的內(nèi)容做轉(zhuǎn)換之后流出,這時(shí)候就需要轉(zhuǎn)換流 Transform。
Transform 流要實(shí)現(xiàn) _transform 的 api,我們實(shí)現(xiàn)下對(duì)內(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)了流動(dòng)。
但是 read 和 write 都是異步的,如果兩者速率不一致呢?
如果 Readable 讀入數(shù)據(jù)的速率大于 Writable 寫入速度的速率,這樣就會(huì)積累一些數(shù)據(jù)在緩沖區(qū),如果緩沖的數(shù)據(jù)過多,就會(huì)爆掉,會(huì)丟失數(shù)據(jù)。
而如果 Readable 讀入數(shù)據(jù)的速率小于 Writable 寫入速度的速率呢?那沒關(guān)系,最多就是中間有段空閑時(shí)期。
這種讀入速率大于寫入速率的現(xiàn)象叫做“背壓”
,或者“負(fù)壓”
。也很好理解,寫入段壓力比較大,寫不進(jìn)去了,會(huì)爆緩沖區(qū),導(dǎo)致數(shù)據(jù)丟失。
這個(gè)緩沖區(qū)大小可以通過 readableHighWaterMark 和 writableHightWaterMark 來查看,是 16k。
怎么解決這種讀寫速率不一致的問題呢?
當(dāng)沒寫完的時(shí)候,暫停讀就行了。這樣就不會(huì)讀入的數(shù)據(jù)越來越多,駐留在緩沖區(qū)。
readable stream 有個(gè) readableFlowing 的屬性,代表是否自動(dòng)讀入數(shù)據(jù),默認(rèn)為 true,也就是自動(dòng)讀入數(shù)據(jù),然后監(jiān)聽 data 事件就可以拿到了。
當(dāng) readableFlowing 設(shè)置為 false 就不會(huì)自動(dòng)讀了,需要手動(dòng)通過 read 來讀入。
readableStream.readableFlowing = false; let data; while((data = readableStream.read()) != null) { console.log(data.toString()); }
但自己手動(dòng) read 比較麻煩,我們依然可以用自動(dòng)流入的方式,調(diào)用 pause 和 resume 來暫停和恢復(fù)就行了。
當(dāng)調(diào)用 writable stream 的 write 方法的時(shí)候會(huì)返回一個(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ù)讀入速率的功能,解決了背壓?jiǎn)栴}。
平時(shí)我們經(jīng)常會(huì)用 pipe 來直接把 Readable 流對(duì)接到 Writable 流,但是好像也沒遇到過背壓?jiǎn)栴},其實(shí)是 pipe 內(nèi)部已經(jīng)做了讀入速率的動(dòng)態(tài)調(diào)節(jié)了。
const rs = fs.createReadStream(src); const ws = fs.createWriteStream(dst); rs.pipe(ws);
流是傳輸數(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 對(duì)象既可以直接調(diào)用 Readable api 創(chuàng)建,然后重寫 _read 方法,也可以繼承 Readable 實(shí)現(xiàn)一個(gè)子類,之后實(shí)例化。其他流同理。(Readable 可以很容易的和 generator 結(jié)合)
當(dāng)讀入的速率大于寫入速率的時(shí)候就會(huì)出現(xiàn)“背壓”現(xiàn)象,會(huì)爆緩沖區(qū)導(dǎo)致數(shù)據(jù)丟失,解決的方式是根據(jù) write 的速率來動(dòng)態(tài) pause 和 resume 可讀流的速率。pipe 就沒有這個(gè)問題,因?yàn)閮?nèi)部做了處理。
流是掌握 IO 繞不過去的一個(gè)概念,而背壓?jiǎn)栴}也是流很常見的問題,遇到了數(shù)據(jù)丟失可以考慮是否發(fā)生了背壓。
感謝你能夠認(rèn)真閱讀完這篇文章,希望小編分享的“Node.js中有幾種stream”這篇文章對(duì)大家有幫助,同時(shí)也希望大家多多支持創(chuàng)新互聯(lián),關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道,更多相關(guān)知識(shí)等著你來學(xué)習(xí)!