解讀Node核心模組Stream系列一(Readable)

連結串列 · 發表 2018-09-18 14:36:45

摘要： node中stream模組是非常，非常，非常重要的一個模組，因為很多模組都是這個模組封裝的： Readable：可讀流，用來讀取資料，比如 fs.createReadStream()。 Writable：可寫流，用來寫資料，比如 fs.createWriteStream()。...

node中stream模組是非常，非常，非常重要的一個模組，因為很多模組都是這個模組封裝的：

Readable：可讀流，用來讀取資料，比如 fs.createReadStream()。
Writable：可寫流，用來寫資料，比如 fs.createWriteStream()。
Duplex：雙工流，可讀+可寫，比如 net.Socket()。
Transform：轉換流，在讀寫的過程中，可以對資料進行修改，比如 zlib.createDeflate()（資料壓縮/解壓）。

node中流的實現：

node中stream是一個類，它繼承自Event模組，所以可以通過事件訂閱的方式來修改內部的狀態或者呼叫外部的回撥，我們可以從原始碼node/lib/internal/streams/legacy.js看到：

node中stream(node/lib/stream.js)包括了主要包括了四個部分：

lib/_stream_readable.js
lib/_stream_writable.js
lib/_stream_tranform.js
lib/_stream_duplex.js

Readble

Readble的例子

客戶端上的 HTTP 響應
伺服器上的 HTTP 請求
fs 讀取的流
zlib 流
crypto 流
TCP socket
子程序 stdout 與 stderr
process.stdin

Readable的特點和簡化實現：

特點

Readable擁有一個通過BufferList生成的快取連結串列buffer，用來快取讀取到的chunk(對於非物件模式的流，資料塊可以是字串或 Buffer。對於物件模式的流，資料塊可是除 null 以外的任意 JavaScript 值)，同時有一個length來記錄buffer的長度
Readable擁有一個highWaterMark來標明buffer的最大容量，通過和length比較決定是否需要補充快取
Readable訂閱'readble'事件來觸發read()消費者從快取中消耗資料
Readable擁有read()從快取區讀取資料的同時也會根據標誌判斷是否呼叫生產者補充快取區
Readable擁有reading來標明消費者正在消耗
Readable擁有howMatchToRead()來隨時調整讀取的大小，防止對buffer過多的讀取，導致會讀取亂碼的部分
Readable擁有fromList()來根據讀取大小的不同，隨時調整buffer中的連結串列結構

由於node原碼的可讀流有將近一千行的程式碼，其中有大量的異常處理，debug除錯，各種可讀流的相容處理，加碼解碼處理等，所以這裡採取一個簡化版的實現，原始碼中使用連結串列作為buffer，這裡採用陣列進行簡化，主要是闡述可讀流的處理過程。

建構函式

Readable擁有一個通過BufferList生成的快取連結串列buffer，用來快取讀取到的chunk(對於非物件模式的流，資料塊可以是字串或 Buffer。對於物件模式的流，資料塊可是除 null 以外的任意 JavaScript 值)，同時有一個length來記錄buffer的長度
Readable擁有一個highWaterMark來標明buffer的最大容量，通過和length比較決定是否需要補充快取
Readable訂閱'readble'事件來觸發read()消費者從快取中消耗資料

const EE = require('events');
const util = require('util');
const fs = require('fs');

function Readable(path,options) {//這個引數是原始碼沒有的，這裡主要是為了讀取fs為案例加的
EE.call(this);//建構函式繼承EventEmiter

this.path = path;
this.autoClose = options.autoClose || true;
this.highWaterMark = options.highWaterMark || 64 * 1024;//64k
this.encoding = options.encoding || null;
this.flags = options.flags || 'r';//// 這個原始碼沒有的，這裡主要是為了fs讀取案例加的
this.needEmitReadable = false;// 需要觸發readable事件，預設不需要
this.position = 0;// 偏移量
this.cache = []; // 快取區
this.reading = false;// 是否正在從快取中讀取，消費者消耗中
this.length = 0; // 快取區大小,控制長度
this.open(); // 這個原始碼沒有的，這裡主要是為了fs讀取案例加的
this.on('newListener', (type) => {
if (type === 'readable') { // 看一下是否是'readable'事件模式
this.read();//消耗者，從buffer讀取資料
}
})
}
util.inherits(Readable, EE);//原型繼承EventEmiter
複製程式碼

下面這個函式在Readable沒有，但是在ReadStream中存在，這裡為了利用fs讀取操作說明流，簡化實現版本上添加了這個方法，後面說明ReadStream模組和Readable的繼承關係

Readable.prototype.open = function(){//這裡是非同步開啟的操作
fs.open(this.path, this.flags, (err, fd) => {
if (err) { // 銷燬檔案
if (this.autoClose) { // 如果需要自動關閉觸發一下銷燬事件
this.destroy(); // 它銷燬檔案
}
return this.emit('error', err);
}
this.fd = fd;
this.emit('open', fd);
});
}
//原始碼中的destory不是這樣的，這裡只是ReadStream中的destory，原始碼中做了各種可讀流的相容組合處理
Readable.prototype.destroy = function() {
if (typeof this.fd != 'number') {
this.emit('close');
} else {
fs.close(this.fd, () => {
this.emit('close');
})
}
}
複製程式碼

read和_read

Readable擁有read()從快取區讀取資料的同時也會根據標誌判斷是否呼叫生產者補充快取區
Readable擁有reading來標明消費者正在消耗
Readable擁有howMatchToRead()來隨時調整讀取的大小，防止對buffer過多的讀取，導致會讀取亂碼的部分
Readable擁有fromList()來根據讀取大小的不同，隨時調整buffer中的連結串列結構

Readable.prototype.read = function(n) {
let buffer = null;
if(n>this.len){// 如果快取區中有資料不夠這次讀取，則調整highWaterMark並且補充快取區
this.highWaterMark = computeNewHighWaterMark(n);//重新計算調整記憶體2的次方 exp: 5 => 8
this.needEmitReadable = true;
this.reading = true;
this._read();
}
if (n > 0 && n < this.len) { // 如果快取區中有資料夠這次讀取，則從快取區中讀取
buffer = Buffer.alloc(n);
let current;
let index = 0;
let flag = true;
//這裡的程式碼就是原始碼中fromList()的功能，對buffer進行調整，exp：123 456 789讀取12 => 3 456 789
while (flag && (current = this.cache.shift())) {//current是一個buffer
for (let i = 0; i < current.length; i++) {
buffer[index++] = current[i];//將快取區中的chunk內容copy到buffer中
if (index == n) {//n個數據讀取完畢
flag = false;
this.length -= n; //快取區長度更新
let c = current.slice(i + 1);//獲取完的chunk exp:123 => 3
if (c.length) { 
this.cache.unshift(c);//將沒有取完的chunk放回 exp: 3
}
break;
}
}
}
}
if(this.length === 0){//快取中沒有資料
this.needEmitReadable = true; 需要觸發'readable'
}
if (this.length < this.highWaterMark) {//快取區沒有滿，補充快取區
this.reading = true;
this._read();
}
return buffer;//read()返回值為一個buffer
}
//第一次讀取是內建的自動讀取到快取區
//然後觸發readable是從快取區中讀取消耗的同時，並且也會補充快取區
Readable.prototype._read = function(n) {
if (typeof this.fd !== 'number') {
return this.once('open', () => this._read());//因為fs.open是非同步函式，當執行read必須要在open之後
}
let buffer = Buffer.alloc(this.highWaterMark);

//原始碼中通過Readable.prototype.push()呼叫readableAddChunk()再呼叫addChunk()
//這裡通過fs.read來呼叫addChunk(this,bytesRead)
fs.read(this.fd, buffer, 0, buffer.length, this.pos, (err, bytesRead) => {
addChunk(this,bytesRead);
})
}

//原始碼中通過Readable.prototype.push()呼叫readableAddChunk()再呼叫addChunk()
function addChunk(stream, chunk) {
stream.length += bytesRead; // 增加快取的個數
stream.position += bytesRead;//記錄檔案讀取的位置
stream.reading = false;
stream.cache.push(buffer);//資料放入到快取中
if (stream.needEmitReadable) {
stream.needEmitReadable = false;
stream.emit('readable');
}
}

//原始碼中這個函式是通過howMatchToRead()呼叫的，因為howMatchToRead()在其他的流中也會用到，所以相容了其他情況
function computeNewHighWaterMark(n) {
n--;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
n++;
return n;
}
複製程式碼

例項

const fs = require('fs');
const rs = fs.createReadStream('./1.txt',{//123456789
flags:'r',
autoClose:true,
highWaterMark:3,
encoding:null
});
rs.on('readable',function () { 
// 如果快取區沒這麼大會返回null
let r = rs.read(1);
console.log(r);
console.log(rs._readableState.length);
rs.read(1);
setTimeout(() => {//因為補充是非同步的
console.log(rs._readableState.length);
}, 1000);
});
複製程式碼

ReadStream

ReadStream和Readable的關係

ReadStream其實是Readable的子類，它繼承了Readable，以fs.createReadStream為例(node/lib/internal/fs/streams.js)：

然後對上面的_read方法進行了覆蓋但是其中呼叫了Readable.prototype.push()方法：

並且在其上擴充套件了open和close：

ReadStream的特點和簡化實現：

特點

ReadStream擁有一個highWaterMark來標明讀取資料的大小
ReadStream訂閱'data'事件來觸發read()消費者讀取資料
ReadStream擁有paused 模式和flowing 模式，它們通過flowing標誌進行控制：

readable.readableFlowing = null，沒有提供消費流資料的機制，所以流不會產生資料。
readable.readableFlowing = true，監聽'data'事件、呼叫readable.pipe()方法、或呼叫readable.resume()方法，會變成true可讀流開始主動地產生資料觸發事件。
readable.readableFlowing = false，呼叫readable.pause()、readable.unpipe()、或接收背壓，會被設為false，暫時停止事件流動但不會停止資料的生成。

Readable擁有read()讀取資料
ReadStream擁有howMatchToRead來隨時調整讀取的大小，防止讀取亂碼

簡化實現

const EE = require('events');
const util = require('util');
const fs = require('fs');
function ReadStream (path,options) {
this.path = path;
this.flags = options.flags || 'r'; //用來標識開啟檔案的模式
this.encoding = options.encoding || null;
this.highWaterMark = options.highWaterMark || 64 * 1024;
this.start = options.start || 0; //讀取(檔案)的開始位置
this.end = options.end || null; //讀取(檔案)的結束位置
this.autoClose = options.autoClose || true;
this.flowing = null; // 預設非流動模式
this.position = this.start // 記錄讀取資料的位置
this.open(); // 開啟文夾
this.on('newListener', function (type) {
if (type === 'data') { // 使用者監聽了data事件
this.flowing = true;
this.read();
}
})
}
ReadStream.prototype.read = function (){
if (typeof this.fd !== 'number') {// open操作是非同步的，所以必須等待檔案開啟this.fd存在說明開啟檔案
return this.once('open', () => this.read());
}
let buffer = Buffer.alloc(this.highWaterMark); // 把資料讀取到這個buffer中
//判斷每次讀取的資料是多少exp:資料來源1234567890 highWaterMark=3
//最後一次讀取長度為1
let howMuchToRead = Math.min(this.end - this.pos + 1, this.highWaterMark);
fs.read(this.fd, buffer, 0, howMuchToRead, this.position, (err, byteRead) => {
if (byteRead > 0) {
this.emit('data', buffer.slice(0, byteRead));
this.position += byteRead;//更新讀取的起點
if (this.flowing) {//處在flowing模式中就一直讀
this.read();
}
}else{//讀取完畢
this.flowing = null;
this.emit('end');
if(this.autoClose){
this.destroy();
}
}
}
//通過flowing控制暫停還是繼續讀取
ReadStream.prototype.pause = function(){
this.flowing = false;
}
ReadStream.prototype.resume = function(){
this.flowing = true;
this.read();
}
ReadStream.prototype.destroy = function () {
if (typeof this.fd != 'number') {
this.emit('close');
} else {
fs.close(this.fd, () => {
this.emit('close');
})
}
};

ReadStream.prototype.open = function() {
fs.open(this.path, this.flags, (err, fd) => {// fd檔案描述符 只要檔案打開了就是number
if (err) {
if (this.autoClose) { // 如果需要自動關閉 觸發一下銷燬事件
this.destroy(); // 銷燬檔案
}
return this.emit('error', err);
}
this.fd = fd;
this.emit('open', fd);
});
};
複製程式碼

例項

let fs = require('fs');
let ReadStream = require('./ReadStream')
let rs = fs.createReadStream('1.txt',{//1234567890
encoding:null, 
flags:'r+', 
highWaterMark:3, 
autoClose:true, 
start:0, 
end:3
});
let arr = [];
rs.on('open',function () {
console.log(' 檔案開啟了')
});
rs.on('data',function (data) {
console.log(data);
arr.push(data);
}); 
rs.on('end',function () { // 只有目標檔案讀取完畢後才觸發
console.log('結束了');
console.log(Buffer.concat(arr).toString());
});
rs.pause()
setTimeout(function () {
rs.resume(); // 恢復的是data事件的觸發
},1000)
rs.on('error',function (err) {
console.log('出錯了')
})
rs.on('close',function () {
console.log('close')
});
複製程式碼

結語：

希望這篇文章能夠讓各位看官對Stream熟悉，因為這個模組是node中的核心，很多模組都是繼承這個模組實現的，如果熟悉了這個模組，對node的使用以及koa等框架的使用將大有好處，接下來會逐步介紹writable等流模式本文參考：