可写流

• 可写流没有会创建，有内容的话会清空
• 默认情况下一次能写 16 * 1024
• 缓存区，第一次写入是真的向文件里写入，第二次在写入的时候放入到了缓存区里
• 写入时候返回一个boolean类型，返回为false时不要在写入
• 当内存和正在写入的内容消耗完后会触发 drain 事件。

读取的时候，读到几个就发射出几个，而可写流默认情况下一次能写入 16 * 1024。举例：读取 30 个，但是写一次写入3个，多出来的 27个则放入到缓存区里。第一次是真往文件里写入，后面则全部放到缓存区中，写完后则清空缓存区。

let fs = require('fs');let ws = fs.createWriteStream('./2.text', {flags: 'w', // 模式encoding: 'utf8', // 编码start: 0, // 从哪儿开始写highWaterMark: 3 // 最高的水位线 16 * 1024
});// 写入 写入的内容必须是字符串或者buffer
// flag 表示是否还有缓存空间 当写入的次数大于等于 highWaterMark 则返回flase
// 可读流配合可写流使用。flag 为了保证读取和写入可以节约内存，而不是疯狂的读
let flag = ws.write('1', 'utf8', () => {
});复制代码

举例子：一次能写3个，9个数字，先写入 9 8 7 然后等一等，等写完后在继续写入 6 5 4 。。。。比如一个人能吃馒头一次最多吃3个，如果不停的给他馒头，嘴里没有吃完则将馒头放到地上（缓存区里去）。等吃完在去拿来吃。

let fs = require('fs');let ws = fs.createWriteStream('./3.text', {flags: 'w', // 模式encoding: 'utf8', // 编码start: 0, // 从哪儿开始写highWaterMark: 3 // 最高的水位线 16 * 1024
});let i = 9;
function write() {// 是否可写入let flag = true;while(flag && i >= 0) {// 第一次 9 8 7 当是 7 的时候 flag 为falseflag = ws.write(i-- + '');}
}
// 只有嘴里塞满了吃完了。才会触发
// drain 触发时机，只有当嘴（highWaterMark）满了,才可能触发 drain
// 只有当嘴里的和地上的（内存里）都吃完了，才会触发 drain 方法
ws.on('drain', () => {console.log('吃完了。');// 继续吃write();
})
write();
复制代码

可写流原理

• 默认属性

constructor(path, options = {}) {super();this.path = path;this.flags = options.flags || 'w';this.encoding = optins.encoding || 'utf8';this.start = options.start || 0;this.mode = options.mode || 0o666;this.autoClose = options.autoClose || true;this.highWaterMark = options.highWaterMark || 16 * 1024;// 先打开文件 异步调用 触发 open 事件后拿到 fdthis.open();
}
复制代码

• open 打开文件

open() {fs.open(this.path, this.flags, this.mode, (err, fd) => {if (err) {// 是否需要关闭if(this.autoClose) {// 如果自动关闭就销毁文件描述符 fdthis.destory();}this.emit('error', err);return;}this.fd = fd;this.emit('open', this.fd);});
}
复制代码

• 关闭文件

// 关闭文件
destory() {if(typeof this.fd !== 'number') {// 没有打开过this.emit('close');} else {fs.close(this.fd, () => {this.emit('close');});}
}
复制代码

• write 写入

// 客户调用的是write方法写入内容
write(chunk, encoding = this.encoding, callback) {// 将 chunk 变成bufferchunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);this.len += chunk.length; // 维护缓存的长度let ret = this.len < this.highWaterMark;if (!ret) {this.needDrain = true; // 需要触发drain事件}if (this.writing) {// 正在写入应该放到内存中this.cach.push({chunk,encoding,callback});} else {// 第一次写入this._write(chunk, encoding, ()=>this.clearBuffer());}return ret; // 表示能不能继续写。false 表示下次写的时候就要占用更多内存
}
复制代码

• _write && clearBuffer 方法实现

_write(chunk, encoding, buffer) {// write 同步调用 fd 还没获取到，获取到fd后在执行writeif (typeof this.fd !== 'number') {return this.once('open', this._write(chunk, encoding, buffer));}fs.write(this.fd, chunk, 0, chunk.length, this.pos, (err, byteWritten) => {this.pos += byteWritten;this.len -= byteWritten; // 每次写入后就要在内存中减少 相当于把馒头吃掉了this.clearBuffer(); // 第一次写完});
}clearBuffer() {let buffer = this.cache.shift();if (buffer) {// 缓存里有this._write(buffer.chunk, buffer.encoding, () => this.clearBuffer());} else {// 缓存里没有// 需要出发drain事件if (this.needDrain) {this.writing = false; // 不需要再写入到内存里。this.needDrain = false;this.emit('drain');}}
}
复制代码

完整代码

<!--WriteStream.js-->
let fs = require('fs');
let EventEmitter = require('events');class WriteStream extends EventEmitter {constructor(path, options = {}) {super();this.path = path;this.flags = options.flags || 'w';this.encoding = options.encoding || 'utf8';this.start = options.start || 0;this.pos = this.start;this.mode = options.mode || 0o666;this.autoClose = options.autoClose || true;this.highWaterMark = options.highWaterMark || 16 * 1024;// 先打开文件 异步调用 触发 open 事件后拿到 fdthis.open();// 写文件的时候需要的参数// 第一次写入是真的往文件里写入this.writing = false; // 默认第一次没有正在写入// 缓存区this.cache = [];// 维护一个变量 表示缓存的长度this.len = 0;// 是否触发 drain 事件this.needDrain = false;}// clearBuffer() {let buffer = this.cache.shift();if (buffer) {// 缓存里有this._write(buffer.chunk, buffer.encoding, () => this.clearBuffer());} else {// 缓存里没有// 需要出发drain事件if (this.needDrain) {this.writing = false; // 不需要再写入到内存里。this.needDrain = false;this.emit('drain');}}}_write(chunk, encoding, clearBuffer) {// write 同步调用 fd 还没获取到，获取到fd后在执行writeif (typeof this.fd !== 'number') {return this.once('open', ()=>this._write(chunk, encoding, clearBuffer));}fs.write(this.fd, chunk, 0, chunk.length, this.pos, (err, byteWritten) => {this.pos += byteWritten;this.len -= byteWritten; // 每次写入后就要在内存中减少 相当于把馒头吃掉了this.clearBuffer(); // 第一次写完});}// 客户调用的是write方法写入内容write(chunk, encoding = this.encoding, callback) {// 将 chunk 变成bufferchunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);this.len += chunk.length; // 维护缓存的长度let ret = this.len < this.highWaterMark;if (!ret) {this.needDrain = true; // 需要触发drain事件}if (this.writing) {// 正在写入应该放到内存中this.cache.push({chunk,encoding,callback});} else {this.writing = true;// 第一次写入this._write(chunk, encoding, ()=>this.clearBuffer());}return ret; // 表示能不能继续写。false 表示下次写的时候就要占用更多内存}// 关闭文件destory() {if(typeof this.fd !== 'number') {// 没有打开过this.emit('close');} else {fs.close(this.fd, () => {this.emit('close');});}}// 打开文件open() {fs.open(this.path, this.flags, this.mode, (err, fd) => {if (err) {// 是否需要关闭if(this.autoClose) {// 如果自动关闭就销毁文件描述符 fdthis.destory();}this.emit('error', err);return;}this.fd = fd;this.emit('open', this.fd);});}}module.exports = WriteStream;<!--应用-->
let fs = require('fs');
let WS = require('./WriteStream');
let ws = new WS('./3.text', {flags: 'w', // 文件如果不存在则创建，文件存在则清空highWaterMark: 3, // 设置当前缓存区大小即嘴有多大encoding: 'utf8', // 文件里存放的是二进制start: 0,autoClose: true, // 自动关闭mode: 0o666 // 可读可写
});let i = 9;function write() {let flag = true;while(flag && i >= 0) {flag = ws.write(i-- + '', 'utf8');}
}
// drain 触发时机，只有当嘴（highWaterMark）满了,才可能触发 drain
// 只有当嘴里的和地上的（内存里）都吃完了，才会触发 drain 方法
write();
ws.on('drain', () => {console.log('写完了');write();
});
复制代码