本文将介绍浅谈Nodejs中的可写流write与实现方法的详细情况，特别是关于nodejs流写入文件的相关信息。我们将通过案例分析、数据研究等多种方式，帮助您更全面地了解这个主题，同时也将涉及一些关于

本文将介绍浅谈Nodejs中的可写流write与实现方法的详细情况，特别是关于nodejs流写入文件的相关信息。我们将通过案例分析、数据研究等多种方式，帮助您更全面地了解这个主题，同时也将涉及一些关于node.js – 在NodeJS中BatchWrite到DynamoDB时的指数退避、node.js中stream流中可读流和可写流的实现与使用方法实例分析、node.js中的buffer.write方法使用说明_node.js、node.js中的fs.createWriteStream方法使用说明_node.js的知识。

• 浅谈Nodejs中的可写流write与实现方法（nodejs流写入文件）
• node.js – 在NodeJS中BatchWrite到DynamoDB时的指数退避
• node.js中stream流中可读流和可写流的实现与使用方法实例分析
• node.js中的buffer.write方法使用说明_node.js
• node.js中的fs.createWriteStream方法使用说明_node.js

浅谈Nodejs中的可写流write与实现方法（nodejs流写入文件）

【推荐学习：《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 表示 当前要写的值是直接是否直接写入文件，不能超出了单次最大写入值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是否有打印

• 效果

自定义可写流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次缓存结果符合预期

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

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

node.js – 在NodeJS中BatchWrite到DynamoDB时的指数退避

// batchwrite to DDB
function batchWriteDDB(params) {
  dynamodb.batchWriteItem(params,function(err,data) {
    if (err) {
      console.error("Batchwrite Failed: " + err,err.stack);
    } else {
      var unprocessed = data.UnprocessedItems;
      if (Object.keys(unprocessed).length === 0) {
        console.log("Processed all items.");
      } else {
        // some unprocessed items,do it again
        console.warn("Batchwrite did not to complete: " + util.inspect(unprocessed,{ showHidden: false,depth: null }));
        console.log("Retry btachwriting...");
        var params2 = {};
        params2["RequestItems"] = data.UnprocessedItems;
        dynamodb.batchWriteItem(params2,function(error,data2) {
          if (err) {
            console.error("Retry Failed: " + err,err.stack);
          } else {
            var unprocessed2 = data2.UnprocessedItems;
            if (Object.keys(unprocessed2).length === 0) {
              console.log("Retry processed all items.");
            } else {
              console.error("Failed AGAIN to complete: " + util.inspect(unprocessed2,depth: null }));
            }
          }
        });
      }
    }
  });
}

AWS.config.update({retryDelayOptions: {base: 300}});
// Delays with maxRetries = 3: 300,600,1200

AWS.config.update({retryDelayOptions: {customBackoff: function(retryCount) {
  // returns delay in ms
}}});

var dynamodb = (new AWS.DynamoDB({maxRetries: 5}))

var dynamodb = new AWS.DynamoDB({maxRetries: 5,retryDelayOptions: {base: 300} });

node.js中stream流中可读流和可写流的实现与使用方法实例分析

本文实例讲述了node.js中stream流中可读流和可写流的实现与使用方法。分享给大家供大家参考，具体如下：

node.js中的流 stream 是处理流式数据的抽象接口。node.js 提供了很多流对象，像http中的request和response，和 process.stdout 都是流的实例。

流可以是 可读的，可写的，或是可读可写的。所有流都是 events 的实例。

一、流的类型

node.js中有四种基本流类型：

1、Writable 可写流 (例：fs.createWriteStream() )

2、Readable 可读流 (例：fs.createReadStream() )

3、Duplex 可读又可写流 (例：net.Socket )

4、Transform 读写过程中可修改或转换数据的 Duplex 流 (例：zlib.createDeflate() )

二、流中的数据有两种模式

1、二进制模式，都是 string字符串  和 Buffer。

2、对象模式，流内部处理的是一系统普通对象。

三、可读流的两种模式

1、流动模式 ( flowing ) ，数据自动从系统底层读取，并通过事件，尽可能快地提供给应用程序。

2、暂停模式 ( paused )，必须显式的调用 read() 读取数据。

可读流 都开始于暂停模式，可以通过如下方法切换到流动模式：

1、添加 ''data'' 事件回调。

2、调用 resume()。

3、调用 pipe()。

可读流通过如下方法切换回暂停模式：

1、如果没有管道目标，调用 pause()。

2、如果有管道目标，移除所有管道目标，调用 unpipe() 移除多个管道目标。

四、创建可读流，并监听事件

const fs = require(''fs'');
//创建一个文件可读流
let rs = fs.createReadStream(''./1.txt'', {
  //文件系统标志
  flags: ''r'',
  //数据编码，如果调置了该参数，则读取的数据会自动解析
  //如果没调置，则读取的数据会是 Buffer
  //也可以通过 rs.setEncoding() 进行设置
  encoding: ''utf8'',
  //文件描述符，默认为null
  fd: null,
  //文件权限
  mode: 0o666,
  //文件读取的开始位置
  start: 0,
  //文件读取的结束位置(包括结束位置)
  end: Infinity,
  //读取缓冲区的大小，默认64K
  highWaterMark: 3
});
//文件被打开时触发
rs.on(''open'', function () {
  console.log(''文件打开'');
});
//监听data事件，会让当前流切换到流动模式
//当流中将数据传给消费者后触发
//由于我们在上面配置了 highWaterMark 为 3字节，所以下面会打印多次。
rs.on(''data'', function (data) {
  console.log(data);
});
//流中没有数据可供消费者时触发
rs.on(''end'', function () {
  console.log(''数据读取完毕'');
});
//读取数据出错时触发
rs.on(''error'', function () {
  console.log(''读取错误'');
});
//当文件被关闭时触发
rs.on(''close'', function () {
  console.log(''文件关闭'');
});

注意，''open'' 和 ''close'' 事件并不是所有流都会触发。

当们监听''data''事件后，系统会尽可能快的读取出数据。但有时候，我们需要暂停一下流的读取，操作其他事情。

这时候就需要用到 pause() 和 resume() 方法。

const fs = require(''fs'');
//创建一个文件可读流
let rs = fs.createReadStream(''./1.txt'', {
  highWaterMark: 3
});
rs.on(''data'', function (data) {
  console.log(`读取了 ${data.length} 字节数据 : ${data.toString()}`);
  //使流动模式的流停止触发''data''事件，切换出流动模式，数据都会保留在内部缓存中。
  rs.pause();
  //等待3秒后，再恢复触发''data''事件，将流切换回流动模式。
  setTimeout(function () {
    rs.resume();
  }, 3000);
});

可读流的 ''readable'' 事件，当流中有数据可供读取时就触发。

注意当监听 ''readable'' 事件后，会导致流停止流动，需调用 read() 方法读取数据。

注意 on(''data'')，on(''readable'')，pipe() 不要混合使用，会导致不明确的行为。

const fs = require(''fs'');
let rs = fs.createReadStream(''./1.txt'', {
  highWaterMark: 1
});
//当流中有数据可供读取时就触发
rs.on(''readable'', function () {
  let data;
  //循环读取数据
  //参数表示要读取的字节数
  //如果可读的数据不足字节数，则返回缓冲区剩余数据
  //如是没有指定字节数，则返回缓冲区中所有数据
  while (data = rs.read()) {
    console.log(`读取到 ${data.length} 字节数据`);
    console.log(data.toString());
  }
});

五、创建可写流，并监听事件

const fs = require(''fs'');
//创建一个文件可写流
let ws = fs.createWriteStream(''./1.txt'', {
  highWaterMark: 3
});
//往流中写入数据
//参数一表示要写入的数据
//参数二表示编码方式
//参数三表示写入成功的回调
//缓冲区满时返回false，未满时返回true。
//由于上面我们设置的缓冲区大小为 3字节，所以到写入第3个时，就返回了false。
console.log(ws.write(''1'', ''utf8''));
console.log(ws.write(''2'', ''utf8''));
console.log(ws.write(''3'', ''utf8''));
console.log(ws.write(''4'', ''utf8''));
function writeData() {
  let cnt = 9;
  return function () {
    let flag = true;
    while (cnt && flag) {
      flag = ws.write(`${cnt}`);
      console.log(''缓冲区中写入的字节数'', ws.writableLength);
      cnt--;
    }
  };
}
let wd = writeData();
wd();
//当缓冲区中的数据满的时候，应停止写入数据，
//一旦缓冲区中的数据写入文件了，并清空了，则会触发 ''drain'' 事件，告诉生产者可以继续写数据了。
ws.on(''drain'', function () {
  console.log(''可以继续写数据了'');
  console.log(''缓冲区中写入的字节数'', ws.writableLength);
  wd();
});
//当流或底层资源关闭时触发
ws.on(''close'', function () {
  console.log(''文件被关闭'');
});
//当写入数据出错时触发
ws.on(''error'', function () {
  console.log(''写入数据错误'');
});

写入流的 end() 方法 和 ''finish'' 事件监听

const fs = require(''fs'');
//创建一个文件可写流
let ws = fs.createWriteStream(''./1.txt'', {
  highWaterMark: 3
});
//往流中写入数据
//参数一表示要写入的数据
//参数二表示编码方式
//参数三表示写入成功的回调
//缓冲区满时返回false，未满时返回true。
//由于上面我们设置的缓冲区大小为 3字节，所以到写入第3个时，就返回了false。
console.log(ws.write(''1'', ''utf8''));
console.log(ws.write(''2'', ''utf8''));
console.log(ws.write(''3'', ''utf8''));
console.log(ws.write(''4'', ''utf8''));
//调用end()表明已经没有数据要被写入，在关闭流之前再写一块数据。
//如果传入了回调函数，则将作为 ''finish'' 事件的回调函数
ws.end(''最后一点数据'', ''utf8'');
//调用 end() 且缓冲区数据都已传给底层系统时触发
ws.on(''finish'', function () {
  console.log(''写入完成'');
});

写入流的 cork() 和 uncork() 方法，主要是为了解决大量小块数据写入时，内部缓冲可能失效，导致的性能下降。

const fs = require(''fs'');
let ws = fs.createWriteStream(''./1.txt'', {
  highWaterMark: 1
});
//调用 cork() 后，会强制把所有写入的数据缓冲到内存中。
//不会因为写入的数据超过了 highWaterMark 的设置而写入到文件中。
ws.cork();
ws.write(''1'');
console.log(ws.writableLength);
ws.write(''2'');
console.log(ws.writableLength);
ws.write(''3'');
console.log(ws.writableLength);
//将调用 cork() 后的缓冲数据都输出到目标，也就是写入文件中。
ws.uncork();

注意 cork() 的调用次数要与 uncork() 一致。

const fs = require(''fs'');
let ws = fs.createWriteStream(''./1.txt'', {
  highWaterMark: 1
});
//调用一次 cork() 就应该写一次 uncork()，两者要一一对应。
ws.cork();
ws.write(''4'');
ws.write(''5'');
ws.cork();
ws.write(''6'');
process.nextTick(function () {
  //注意这里只调用了一次 uncork()
  ws.uncork();
  //只有调用同样次数的 uncork() 数据才会被输出。
  ws.uncork();
});

六、可读流的 pipe() 方法

pipe() 方法类似下面的代码，在可读流与可写流之前架起一座桥梁。

const fs = require(''fs'');
//创建一个可读流
let rs = fs.createReadStream(''./1.txt'', {
  highWaterMark: 3
});
//创建一个可写流
let ws = fs.createWriteStream(''./2.txt'', {
  highWaterMark: 3
});
rs.on(''data'', function (data) {
  let flag = ws.write(data);
  console.log(`往可写流中写入 ${data.length} 字节数据`);
  //如果写入缓冲区已满，则暂停可读流的读取
  if (!flag) {
    rs.pause();
    console.log(''暂停可读流'');
  }
});
//监控可读流数据是否读完
rs.on(''end'', function () {
  console.log(''数据已读完'');
  //如果可读流读完了，则调用 end() 表示可写流已写入完成
  ws.end();
});
//如果可写流缓冲区已清空，可以再次写入，则重新打开可读流
ws.on(''drain'', function () {
  rs.resume();
  console.log(''重新开启可读流'');
});

我们用 pipe() 方法完成上面的功能。

const fs = require(''fs'');
//创建一个可读流
let rs = fs.createReadStream(''./1.txt'', {
  highWaterMark: 3
});
//创建一个可写流
let ws = fs.createWriteStream(''./2.txt'', {
  highWaterMark: 3
});
let ws2 = fs.createWriteStream(''./3.txt'', {
  highWaterMark: 3
});
//绑定可写流到可读流，自动将可读流切换到流动模式，将可读流的所有数据推送到可写流。
rs.pipe(ws);
//可以绑定多个可写流
rs.pipe(ws2);

我们也可以用 unpipe() 手动的解绑可写流。

const fs = require(''fs'');
//创建一个可读流
let rs = fs.createReadStream(''./1.txt'', {
  highWaterMark: 3
});
//创建一个可写流
let ws = fs.createWriteStream(''./2.txt'', {
  highWaterMark: 3
});
let ws2 = fs.createWriteStream(''./3.txt'', {
  highWaterMark: 3
});
rs.pipe(ws);
rs.pipe(ws2);
//解绑可写流，如果参数没写，则解绑所有管道
setTimeout(function () {
  rs.unpipe(ws2);
}, 0);

希望本文所述对大家node.js程序设计有所帮助。

node.js中的buffer.write方法使用说明_node.js

方法说明：

将string使用指定的encoding写入到buffer的offset处。

返回写入了多少个八进制字节。

如果Buffer没有足够的空间来适应整个string,那么将只有string的部分被写入。

语法：