Pro
一個createWriteStream的簡單實現(xiàn)牙甫，以求能增加對可寫流的理解與應(yīng)用滩褥。
參數(shù)配置
/**
* createWriteStream
* @param1 path
* @param2 options
*/
let fs = require('fs');
let ws = fs.createWriteStream('./1.txt',{
flags:'w'//文件的打開模式
,mode:0o666//文件的權(quán)限設(shè)置
,encoding:'utf8'//寫入文件的字符的編碼
,highWaterMark:3//最高水位線
,start:0 //寫入文件的起始索引位置
,autoClose:true//是否自動關(guān)閉文檔
})
createWriteStream類的實例化
- 實例化一個
createWriteStream類- 將
path，options掛載在createWriteStream的實例上炫加，除此之外再在實例上掛載以下屬性-
self.fd=null：文件打開后返回的文件描述符 -
self.pos=self.start：用于表示文件真正寫入時的指針位置 -
self.Buffer=[]：用來表示文件的緩沖區(qū) -
self.len=null：用來表示緩沖區(qū)此時的大小 -
self.isWriting=false：用來表示是否正在真正寫入文件
-
- 調(diào)用
open方法瑰煎，打開文件(發(fā)射open事件)
- 將
實例write方法的執(zhí)行流程
-
wirte方法接收三個參數(shù)，chunk要寫入的內(nèi)容俗孝，encoding要進行的酒甸，cb回調(diào)函數(shù)。 -
write執(zhí)行流程：- 判斷傳入的
chunk是否為buffer赋铝，如果不是插勤，則轉(zhuǎn)換成buffer，用于轉(zhuǎn)化編碼依據(jù)傳入的encoding參數(shù)革骨。 - 更新
Buffer緩沖區(qū)的len長度农尖，讓len加上該次chunk的長度 - 判斷
len是否已經(jīng)超過highWaterMark，將值存入flag - 判斷是否處于
isWriting狀態(tài)：- 是良哲，則先加
chunk寫入實例對象下的Buffer緩沖區(qū)盛卡。 - 否，更新
isWriting筑凫，接將參數(shù)傳遞給實例下的_write方法寫入文件
- 是良哲，則先加
- 返回
flag
- 判斷傳入的
實例_write方法的執(zhí)行流程
此方法用于真正寫入文件
- 查看實例的
fd屬性是否存在(文件是否打開成功)- 成功滑沧，調(diào)用
fs模塊的write方法正式寫入數(shù)據(jù)- 更新實例對象下的
len以及pos屬性 - 調(diào)用
clearBuffer方法將緩沖區(qū)的內(nèi)容寫入 - 調(diào)用write方法傳入的回調(diào)函數(shù)
cb
- 更新實例對象下的
- 失敗，訂閱一個
open事件（open事件將會在open方法中被發(fā)射）巍实，在訂閱中的回調(diào)方法中再次以相同的參數(shù)調(diào)用_write方法
- 成功滑沧，調(diào)用
實例clearBuffer方法
- 從緩沖區(qū)中取出一個數(shù)據(jù)
- 如果數(shù)據(jù)存在滓技，調(diào)用
_write方法 - 如果數(shù)據(jù)不存在，將
isWriting更改為false棚潦，發(fā)射drain事件
- 如果數(shù)據(jù)存在滓技，調(diào)用
實現(xiàn)源碼以及測試文件
let fs = require('fs');
let EventEmitter = require('events');
class WriteStream extends EventEmitter {
constructor(path, options) {
super();
let self = this;
Object.assign(self, options); //還需設(shè)置默認值
self.path = path;
self.isWriting = false;
self.Buffer = []; //源碼中為鏈表實現(xiàn)的緩沖區(qū)
self.len = null;
self.pos = self.start; //初始化寫入位置
self.fd = null;
self.open();
}
open() {
let self = this;
fs.open(self.path, self.flags, self.mode, (err, fd) => {
self.fd = fd;
if (err) return self.destroy(err);
self.emit('open');
});
}
destroy(err) {
fs.close(this.fd, () => {
this.emit('error', err);
});
}
write(chunk, encoding, cb) {
let self = this
, ret = null;
encoding = encoding?encoding:self.encoding; //優(yōu)先使用write傳入的編碼方式
chunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding);
self.len += chunk.length;
ret = self.highWaterMark > self.len; //判斷當前最新的緩沖區(qū)是否已達到最高水位線
if (self.isWriting) { //說明正在調(diào)用底層方法真正寫入文件令漂，先寫入Buffer
self.Buffer.push({
chunk
, cb
});
} else {
self.isWriting = true;
self._write(chunk, cb, () => self.clearBuffer());
}
return ret;
}
_write(chunk, cb, clear) {
let self = this;
if (!self.fd) return self.once('open', () => {
self._write(chunk, cb, clear)
});
fs.write(self.fd, chunk, 0, chunk.length, self.pos, (err, bytesWritten) => {
if (err) {
if (self.autoClose) {
self.destroy();
self.emit('error', err);
}
}
self.len -= bytesWritten;
self.pos += bytesWritten;
cb && cb();
clear && clear();
});
}
clearBuffer() {
let self = this
, data = null;
data = self.Buffer.shift();
if (data) {
self._write(data.chunk, data.cb, () => self.clearBuffer());
} else { //此時說明緩沖區(qū)已無數(shù)據(jù)
self.isWriting = false;
self.emit('drain');
}
}
}
module.exports = WriteStream;
測試文件:
let WriteStream = require('./practice');
let ws = new WriteStream('./1.txt',{
flags:'w'
,mode:0o666
,start:0
,encoding:'utf8'
,autoClose:true //當流寫完之后自動關(guān)閉文件
,highWaterMark:3
});
let n = 9;
ws.on('error',(err)=>{
console.log(err)
})
function write(){
let flag = true;
while(flag&&n>0){
flag = ws.write(n+"",'utf8',()=>{
console.log('ok');
});
n--;
console.log('flag=',flag)
}
ws.once('drain',()=>{
console.log('drain');
write();
})
}
// ws.on('drain',()=>{
// console.log('drain');
// write();
// })
write();
參考資料:
https://nodejs.org/dist/latest-v9.x/docs/api/stream.html#stream_writable_streams
