浅谈Nodejs中的可写流write与实现方法

【推荐学习：《nodejs 教程》】

可写流-Writable

fs.createWritestream调用例子

• 首次读取的数据会真实写入目标文件
• 其余次读取的数据要根据读取数据是否超出highWaterMark ，是的话存入缓存区等待写入目标文件中

const fs = require("fs");
    const path = require("path");
    const bPath = path.join(__dirname, "b.txt");
let ws = fs.createWriteStream(bPath, {
  flags: "w",  encoding: "utf-8",  autoClose: true,  start: 0,  highWaterMark: 3,}
    );
ws.on("oPEn", function (fd) {
      console.LOG("open", fd);
}
    );
ws.on("close", function () {
      console.log("close");
}
    );
     //string 或者buffer，ws.write 还有一个boolea的返回值ws.write("1");
    //flag 表示 当前要写的值是直接是否直接写入文件，不能超出了单次最大写入值highWaterMarklet flag = ws.write("1");
console.log({
 flag }
    );
    //trueflag = ws.write("1");
console.log({
 flag }
    );
    //falseflag = ws.write("1");
console.log({
 flag }
    );
    //falseflag = ws.write("14444444");
console.log({
 flag }
    );
    //falsews.end();
     //write+close,没有调用 end 是不会调用 触发close的，看到这里的小伙伴可以尝试注释end() 看看close的console是否有打印

• 效果

自定义可写流initWriteStream

继承EventEmitter发布订阅

const EventEmitter = require("events");
    const fs = require("fs");
class WriteStream extends EventEmitter {
}
    module.exports = WriteStream;
    

链表生成队列做文件读取的缓存

链表& 队列的实现

https://juejin.cn/post/6973847774752145445

// 用链表 生成队列 对 文件缓存区的读取 进行优化const Queue = require("./queue");
    

初始化实例默认数据constructor()

 constructor(path, options = {
}
) {
        super();
        this.path = path;
        this.flags = options.flags || "w";
        this.encoding = options.encoding || "utf8";
        this.mode = options.mode || 0o666;
     //默认8进制 ，6 6 6  三组分别的权限是 可读可写    this.autoClose = options.start || 0;
        this.highWaterMark = options.highWaterMark || 16 * 1024;
     //默认一次读取16个字节的数据    this.len = 0;
     //用于维持有多少数据还没有被写入文件中    //是否根据等待当前读取的最大文数据 排空后再写入    this.needDrain = false;
     //    // 缓存队列 用于存放 非第一次的文件读取 到的数据，因为第一次读取 直接塞入目标文件中    // 除第一次 的文件读取数据的都存放再缓存中    // this.cache = [];
        // 队列做缓存    this.cache = new Queue();
        // 标记是否是第一次写入目标文件的标识    this.writing = false;
        this.start = options.start || 0;
        this.offset = this.start;
     //偏移量    this.open();
  }
    

• this.mode 文件操作权限 默认0o666(0o表示8进制)

  • 3个6所占位置分别对应：文件所属用户对它的权限 ；文件所属用户组用户对它的权限；表示其他用户对它的权限

  • 权限由：r--可读（对应数值4），w--可写(对应数值2)，x--可执行(对应数值1，例如文件夹下有 .exe 这样的标识 说明点击可以直接执行)组成

  • 所以默认情况下3组用户对文件的操作权限都是可读可写

open()

• 调用fs.open()
• 回调emit实例open方法，fs.open的返回值fd做参数传入

 open() {
        fs.open(this.path, this.flags, this.mode, (err, fd) =>
 {
          this.fd = fd;
          this.emit("open", fd);
    }
    );
  }
    

write()

• 转化实例传入的需要写入的文件数据格式为buffer
• 判断写入数据长度是否大于highWaterMark，如果达到预期后，文件读取到的数据存放再缓存里 不直接写入目标文件（这里要排除是否是第一次读取文件）
• 执行实例write 传入的cb 并调用clearBuffer 清空缓存
• 判断 是否是第一次读取，第一次读取 直接写入调用 _write(待实现)
• 缓存队列尾部offer 当前读取到的数据等待写入目标文件

 write(chunk, encoding = this.encoding, cb = () =>
 {
}
) {
        //  将数据全部转换成buffer    chunk = Buffer.isBuffer(chunk) ? chunk : Buffer.From(chunk);
        this.len += chunk.length;
    // console.log({
chunk}
    ,this.len )    let returnValue = this.len  this.highWaterMark;
        //当数据写入后，需要在手动的将this.len--    this.needDrain = !returnValue;
     //如果达到预期 后 的文件读取 到数据存放再缓存里 不直接写入目标文件    //清空缓存 对用户传入的回调 进行二次包装    let userCb = cb;
        cb = () =>
 {
          userCb();
          //清空buffer      this.clearBuffer();
//马上实现    }
    ;
    //此时需要判断 是否是第一次读取，第一次读取 直接写入调用 _write    if (!this.writing) {
          // 第一次||缓存队列已清空完毕      this.writing = true;
          // console.log("First write");
          this._write(chunk, encoding, cb);
//马上实现    }
 else {
    //缓存队列尾部offer 当前读取到的数据等待写入目标文件      this.cache.offer({
        chunk,        encoding,        cb,      }
    );
    }
        return returnValue;
  }
    

clearBuffer()依次清空缓存队列

• 队列执行顺序，先进先出原则
• this.cache.poll() 依次拿取头部数据执行this._write写入目标文件
• 缓存队列poll出来的data如果不存在，则说明是第一次写入的行为||缓存队列已清空。this.writing = false; 下次的文件读取可以直接写入目标文件
• 如果this.needDrain又达到预期，文件读取到数据存放再缓存里 不直接写入目标文件

clearBuffer() {
        //写入成功后 调用 clearBuffer--》写入缓存第一个，第一个完成后，再继续 第二个    let data = this.cache.poll();
    // console.log('this.cache',this.cache)    if (data) {
          //有值 写入文件      this._write(data.chunk, data.encoding, data.cb);
    }
 else {
          this.writing = false;
      if (this.needDrain) {
            // 如果是缓存，触发drain        this.emit("drain");
      }
    }
  }
    

_write()

• fs.open()是异步的，成功读取后fd会是一个number类型
• 根据fd的type 决定是否订阅一次open，并回调自己（直到fd类型为number）
• fd类型为number：调用fs.write，写入当前的chunk，

 _write(chunk, encoding, cb) {
    if (typeof this.fd !== "number") {
          return this.once("open", () =>
     this._write(chunk, encoding, cb));
    }
        fs.write(this.fd, chunk, 0, chunk.length, this.offset, (err, written) =>
 {
          this.offset += written;
     //维护偏移量      this.len -= written;
     //把缓存的个数减少      cb();
     //写入成功      // console.log(this.cache);
    }
    );
  }
    

测试自定义的Writable

const WriteStream = require("./initWriteStream");
let ws = new WriteStream(bPath, {
  highWaterMark: 3,}
    );
    let i = 0;
function write() {
      //写入0-9个  let flag = true;
      while (i  10 &
    &
 flag) {
        flag = ws.write(i++ + "");
         console.log(flag);
  }
}
ws.on("drain", function () {
      // 只有当我们写入的数据达到预期，并且数据被清空后才会触发drain ⌚️  console.log("写完了");
      write();
}
    );
    write();
    

• 10个数字，依次写入，3次达到最大预期值，然后依次清空了3次缓存结果符合预期

• 目标文件中查看是否正确写入了我们预期的数值

更多编程相关知识，请访问：编程视频！！

以上就是浅谈Nodejs中的可写流write与实现方法的详细内容，更多请关注其它相关文章！

声明：本文内容由网友自发贡献，本站不承担相应法律责任。对本内容有异议或投诉，请联系2913721942#qq.com核实处理，我们将尽快回复您，谢谢合作！