💎一站式轻松地调用各大LLM模型接口,支持GPT4、智谱、星火、月之暗面及文生图 广告
## Pro 一个`createWriteStream`的简单实现,以求能增加对可写流的理解与应用。 ## 参数配置 ``` /** * createWriteStream * @param1 path * @param2 options */ let fs = require('fs'); let ws = fs.createWriteStream('./1.txt',{ flags:'w'//文件的打开模式 ,mode:0o666//文件的权限设置 ,encoding:'utf8'//写入文件的字符的编码 ,highWaterMark:3//最高水位线 ,start:0 //写入文件的起始索引位置 ,autoClose:true//是否自动关闭文档 }) ``` ## createWriteStream类的实例化 - 实例化一个`createWriteStream`类 - 将`path`,`options`挂载在`createWriteStream`的实例上,除此之外再在实例上挂载以下属性 - `self.fd=null`:文件打开后返回的文件描述符 - `self.pos=self.start`:用于表示文件真正写入时的指针位置 - `self.Buffer=[]`:用来表示文件的缓冲区 - `self.len=null`:用来表示缓冲区此时的大小 - `self.isWriting=false`:用来表示是否正在真正写入文件 - 调用`open`方法,打开文件(发射open事件) ## 实例write方法的执行流程 - `wirte`方法接收三个参数,`chunk`要写入的内容,`encoding`要进行的,`cb`回调函数。 - `write`执行流程: - 判断传入的`chunk`是否为buffer,如果不是,则转换成buffer,用于转化编码依据传入的`encoding`参数。 - 更新`Buffer`缓冲区的`len`长度,让len加上该次chunk的长度 - 判断`len`是否已经超过`highWaterMark`,将值存入`flag` - 判断是否处于`isWriting`状态: - 是,则先加`chunk`写入实例对象下的`Buffer缓冲区`。 - 否,更新`isWriting`,接将参数传递给实例下的`_write`方法写入文件 - 返回`flag` ## 实例_write方法的执行流程 此方法用于真正写入文件 - 查看实例的`fd`属性是否存在(文件是否打开成功) - 成功,调用`fs`模块的`write`方法正式写入数据 - 更新实例对象下的`len`以及`pos`属性 - 调用`clearBuffer`方法将缓冲区的内容写入 - 调用write方法传入的回调函数`cb` - 失败,订阅一个`open事件`(open事件将会在open方法中被发射),在订阅中的回调方法中再次以相同的参数调用_write方法 ## 实例clearBuffer方法 - 从缓冲区中取出一个数据 - 如果数据存在,调用`_write`方法 - 如果数据不存在,将`isWriting`更改为false,发射`drain`事件 ## 注意事项 - 若写入没有暂停过,即写入本地文件的速度远大于write()调用的速度,那么不会走写入缓存,也就不会触发`drain`事件。 ## 实现源码以及测试文件 ``` let fs = require('fs'); let EventEmitter = require('events'); class WriteStream extends EventEmitter { constructor(path, options) { super(); let self = this; Object.assign(self, options); //还需设置默认值 self.path = path; self.isWriting = false; self.Buffer = []; //源码中为链表实现的缓冲区 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; //优先使用write传入的编码方式 chunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding); self.len += chunk.length; ret = self.highWaterMark > self.len; //判断当前最新的缓冲区是否已达到最高水位线 if (self.isWriting) { //说明正在调用底层方法真正写入文件,先写入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 { //此时说明缓冲区已无数据 self.isWriting = false; self.emit('drain'); } } } module.exports = WriteStream; ``` 测试文件: ```// let fs = require('fs'); let WriteStream = require('./practice'); let ws = new WriteStream('./1.txt',{ flags:'w' ,mode:0o666 ,start:0 ,encoding:'utf8' ,autoClose:true //当流写完之后自动关闭文件 ,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](https://nodejs.org/dist/latest-v9.x/docs/api/stream.html#stream_writable_streams)