Home > Article > Web Front-end > A brief discussion on the writable stream write and implementation methods in Nodejs
This article will take you to understand the writable stream write in Nodejs, and introduce the implementation of Node's writable stream write. It has certain reference value. Friends in need can refer to it. I hope it will be helpful to everyone.
[Recommended learning: "nodejs tutorial"]
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 表示 当前要写的值是直接是否直接写入文件,不能超出了单次最大写入值highWaterMark let flag = ws.write("1"); console.log({ flag });//true flag = ws.write("1"); console.log({ flag });//false flag = ws.write("1"); console.log({ flag });//false flag = ws.write("14444444"); console.log({ flag });//false ws.end(); //write+close,没有调用 end 是不会调用 触发close的,看到这里的小伙伴可以尝试注释end() 看看close的console是否有打印
const EventEmitter = require("events"); const fs = require("fs"); class WriteStream extends EventEmitter {} module.exports = WriteStream;
Implementation of linked list & queue
https://juejin.cn/post/6973847774752145445
// 用链表 生成队列 对 文件缓存区的读取 进行优化 const Queue = require("./queue");
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 file operation permission defaults to 0o666 (0o represents octal)
The positions occupied by the three 6s are respectively Corresponding to: the permissions of the user to whom the file belongs; the permissions of the user group to which the file belongs; indicating the permissions of other users on it
Permissions are represented by: r--readable (corresponding value 4), w--writable (corresponding to value 2), x--executable (corresponding to value 1, for example, if there is an .exe mark under the folder, it means that clicking can be executed directly) to form
So by default, the operating permissions of the three groups of users on the file are readable and writable
open() { fs.open(this.path, this.flags, this.mode, (err, fd) => { this.fd = fd; this.emit("open", fd); }); }
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() { //写入成功后 调用 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(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); }); }
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();
For more programming-related knowledge, please visit: programming video! !
The above is the detailed content of A brief discussion on the writable stream write and implementation methods in Nodejs. For more information, please follow other related articles on the PHP Chinese website!