1. Node.js 中有四種基本的流類型
- Readable - 可讀的流 (例如 fs.createReadStream()).
- Writable - 可寫的流 (例如 fs.createWriteStream()).
- Duplex - 可讀寫的流 (例如 net.Socket).
- Transform - 在讀寫過程中可以修改和變換數(shù)據(jù)的 Duplex 流 (例如 zlib.createDeflate()).
2. 流中的數(shù)據(jù)有兩種模式,二進(jìn)制模式和對象模式
二進(jìn)制模式, 每個(gè)分塊都是buffer或者string對象.
-
對象模式, 流內(nèi)部處理的是一系列普通對象.
所有使用 Node.js API 創(chuàng)建的流對象都只能操作 strings 和 Buffer對象愤兵。但是侠驯，通過一些第三方流的實(shí)現(xiàn)秋度，你依然能夠處理其它類型的 JavaScript 值 (除了 null，它在流處理中有特殊意義)。 這些流被認(rèn)為是工作在 “對象模式”（object mode）薯演。 在創(chuàng)建流的實(shí)例時(shí)贸诚，可以通過 objectMode 選項(xiàng)使流的實(shí)例切換到對象模式。試圖將已經(jīng)存在的流切換到對象模式是不安全的销钝。
3. 可讀流的兩種模式
- 可讀流事實(shí)上工作在下面兩種模式之一：
flowing和paused - 在 flowing 模式下有咨， 可讀流自動(dòng)從系統(tǒng)底層讀取數(shù)據(jù)，并通過 EventEmitter 接口的事件盡快將數(shù)據(jù)提供給應(yīng)用曙搬。
- 在 paused 模式下摔吏，必須顯式調(diào)用 stream.read() 方法來從流中讀取數(shù)據(jù)片段。
- 所有初始工作模式為 paused 的 Readable 流纵装，可以通過下面三種途徑切換到 flowing 模式：
- 監(jiān)聽 'data' 事件
- 調(diào)用 stream.resume() 方法
- 調(diào)用 stream.pipe() 方法將數(shù)據(jù)發(fā)送到 Writable
- 可讀流可以通過下面途徑切換到 paused 模式：
- 如果不存在管道目標(biāo)（pipe destination）征讲，可以通過調(diào)用 stream.pause() 方法實(shí)現(xiàn)。
- 如果存在管道目標(biāo)橡娄，可以通過取消 'data' 事件監(jiān)聽诗箍，并調(diào)用 stream.unpipe() 方法移除所有管道目標(biāo)來實(shí)現(xiàn)。
如果 Readable 切換到 flowing 模式挽唉，且沒有消費(fèi)者處理流中的數(shù)據(jù)滤祖，這些數(shù)據(jù)將會(huì)丟失。 比如瓶籽， 調(diào)用了 readable.resume() 方法卻沒有監(jiān)聽 'data' 事件匠童，或是取消了 'data' 事件監(jiān)聽，就有可能出現(xiàn)這種情況塑顺。
4.緩存區(qū)
Writable 和 Readable 流都會(huì)將數(shù)據(jù)存儲(chǔ)到內(nèi)部的緩沖器（buffer）中汤求。這些緩沖器可以 通過相應(yīng)的 writable._writableState.getBuffer() 或 readable._readableState.buffer 來獲取。
緩沖器的大小取決于傳遞給流構(gòu)造函數(shù)的 highWaterMark 選項(xiàng)严拒。 對于普通的流扬绪， highWaterMark 選項(xiàng)指定了總共的字節(jié)數(shù)。對于工作在對象模式的流裤唠， highWaterMark 指定了對象的總數(shù)挤牛。
當(dāng)可讀流的實(shí)現(xiàn)調(diào)用
stream.push(chunk)方法時(shí)，數(shù)據(jù)被放到緩沖器中种蘸。如果流的消費(fèi)者沒有調(diào)用stream.read()方法墓赴， 這些數(shù)據(jù)會(huì)始終存在于內(nèi)部隊(duì)列中竞膳，直到被消費(fèi)。當(dāng)內(nèi)部可讀緩沖器的大小達(dá)到 highWaterMark 指定的閾值時(shí)竣蹦，流會(huì)暫停從底層資源讀取數(shù)據(jù)顶猜，直到當(dāng)前 緩沖器的數(shù)據(jù)被消費(fèi) (也就是說， 流會(huì)在內(nèi)部停止調(diào)用 readable._read() 來填充可讀緩沖器)痘括。
可寫流通過反復(fù)調(diào)用 writable.write(chunk) 方法將數(shù)據(jù)放到緩沖器长窄。 當(dāng)內(nèi)部可寫緩沖器的總大小小于 highWaterMark 指定的閾值時(shí)， 調(diào)用 writable.write() 將返回true纲菌。 一旦內(nèi)部緩沖器的大小達(dá)到或超過 highWaterMark 挠日，調(diào)用 writable.write() 將返回 false 。
stream API 的關(guān)鍵目標(biāo)翰舌， 尤其對于 stream.pipe() 方法嚣潜， 就是限制緩沖器數(shù)據(jù)大小，以達(dá)到可接受的程度椅贱。這樣懂算，對于讀寫速度不匹配的源頭和目標(biāo)，就不會(huì)超出可用的內(nèi)存大小庇麦。
Duplex 和 Transform 都是可讀寫的计技。 在內(nèi)部，它們都維護(hù)了 兩個(gè) 相互獨(dú)立的緩沖器用于讀和寫山橄。 在維持了合理高效的數(shù)據(jù)流的同時(shí)垮媒，也使得對于讀和寫可以獨(dú)立進(jìn)行而互不影響。
5. 可讀流的三種狀態(tài)
在任意時(shí)刻航棱，任意可讀流應(yīng)確切處于下面三種狀態(tài)之一：
readable._readableState.flowing = null
readable._readableState.flowing = false
readable._readableState.flowing = true
若 readable._readableState.flowing 為 null睡雇，由于不存在數(shù)據(jù)消費(fèi)者，可讀流將不會(huì)產(chǎn)生數(shù)據(jù)饮醇。 在這個(gè)狀態(tài)下它抱，監(jiān)聽 'data' 事件，調(diào)用 readable.pipe() 方法朴艰，或者調(diào)用 readable.resume() 方法观蓄， readable._readableState.flowing 的值將會(huì)變?yōu)?true 。這時(shí)呵晚，隨著數(shù)據(jù)生成，可讀流開始頻繁觸發(fā)事件沫屡。
調(diào)用 readable.pause() 方法饵隙， readable.unpipe() 方法， 或者接收 “背壓”（back pressure）沮脖， 將導(dǎo)致 readable._readableState.flowing 值變?yōu)?false金矛。 這將暫停事件流芯急，但 不會(huì) 暫停數(shù)據(jù)生成。 在這種情況下驶俊，為 'data' 事件設(shè)置監(jiān)聽函數(shù)不會(huì)導(dǎo)致 readable._readableState.flowing 變?yōu)?true娶耍。
當(dāng) readable._readableState.flowing 值為 false 時(shí)， 數(shù)據(jù)可能堆積到流的內(nèi)部緩存中饼酿。
6.readable
'readable' 事件將在流中有數(shù)據(jù)可供讀取時(shí)觸發(fā)榕酒。在某些情況下，為 'readable' 事件添加回調(diào)將會(huì)導(dǎo)致一些數(shù)據(jù)被讀取到內(nèi)部緩存中故俐。
const readable = getReadableStreamSomehow();
readable.on('readable', () => {
// 有一些數(shù)據(jù)可讀了
});
當(dāng)?shù)竭_(dá)流數(shù)據(jù)尾部時(shí)想鹰， 'readable' 事件也會(huì)觸發(fā)。觸發(fā)順序在 'end' 事件之前药版。
-
事實(shí)上辑舷， 'readable' 事件表明流有了新的動(dòng)態(tài)：要么是有了新的數(shù)據(jù)，要么是到了流的尾部槽片。 對于前者何缓， stream.read() 將返回可用的數(shù)據(jù)。而對于后者还栓， stream.read() 將返回 null碌廓。
let fs =require('fs'); let rs = fs.createReadStream('./1.txt',{ start:3, end:8, encoding:'utf8', highWaterMark:3 }); rs.on('readable',function () { console.log('readable'); console.log('rs._readableState.buffer.length',rs._readableState.length); let d = rs.read(1); console.log('rs._readableState.buffer.length',rs._readableState.length); console.log(d); setTimeout(()=>{ console.log('rs._readableState.buffer.length',rs._readableState.length); },500) });
7.流的經(jīng)典應(yīng)用
7.1 行讀取器
7.1.1 換行和回車
- 以前的打印要每秒可以打印10個(gè)字符，換行城要0.2秒蝙云，正要可以打印2個(gè)字符氓皱。
- 研制人員就是在每行后面加兩個(gè)表示結(jié)束的字符。一個(gè)叫做"回車"勃刨，告訴打字機(jī)把打印頭定位在左邊界波材；另一個(gè)叫做"換行"，告訴打字機(jī)把紙向下移一行身隐。
- Unix系統(tǒng)里廷区，每行結(jié)尾只有換行"(line feed)"，即"\n",
- Windows系統(tǒng)里面贾铝，每行結(jié)尾是"<回車><換行>"隙轻，即"\r\n"
- Mac系統(tǒng)里，每行結(jié)尾是"回車"(carriage return)垢揩，即"\r"
- 在ASCII碼里
- 換行 \n 10 0A
- 回車 \r 13 0D
7.1.2 代碼
let fs = require('fs');
let EventEmitter = require('events');
let util = require('util');
util.inherits(LineReader, EventEmitter)
fs.readFile('./1.txt',function (err,data) {
console.log(data);
})
function LineReader(path) {
EventEmitter.call(this);
this._rs = fs.createReadStream(path);
this.RETURN = 0x0D;// \r 13
this.NEW_LINE = 0x0A;// \n 10
this.on('newListener', function (type, listener) {
if (type == 'newLine') {
let buffer = [];
this._rs.on('readable', () => {
let bytes;
while (null != (bytes = this._rs.read(1))) {
let ch = bytes[0];
switch (ch) {
case this.RETURN:
this.emit('newLine', Buffer.from(buffer));
buffer.length = 0;
let nByte = this._rs.read(1);
if (nByte && nByte[0] != this.NEW_LINE) {
buffer.push(nByte[0]);
}
break;
case this.NEW_LINE:
this.emit('newLine', Buffer.from(buffer));
buffer.length = 0;
break;
default:
buffer.push(bytes[0]);
break;
}
}
});
this._rs.on('end', () => {
if (buffer.length > 0) {
this.emit('newLine', Buffer.from(buffer));
buffer.length = 0;
this.emit('end');
}
})
}
});
}
var lineReader = new LineReader('./1.txt');
lineReader.on('newLine', function (data) {
console.log(data.toString());
}).on('end', function () {
console.log("end");
})
