ホームページ > 記事 > ウェブフロントエンド > Nodejs での書き込み可能なストリームの書き込みと実装方法に関する簡単な説明

Nodejs での書き込み可能なストリームの書き込みと実装方法に関する簡単な説明

青灯夜游転載: 2021-06-21 10:08:542139ブラウズ

この記事では、Nodejs での書き込み可能なストリーム書き込みを理解し、Node の書き込み可能なストリーム書き込みの実装を紹介します。一定の参考値があるので、困っている友達が参考になれば幸いです。

[推奨学習:「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是否有打印

Effect

Nodejs での書き込み可能なストリームの書き込みと実装方法に関する簡単な説明

カスタマイズされた書き込み可能ストリーム 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(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 が占める位置は、それぞれ、ファイルが属するユーザーの権限、ファイルが属するユーザーグループの権限、そのファイルに対する他のユーザーの権限を示します。

Call fs.open()

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

write ()

変換インスタンスによって渡される、書き込む必要があるファイルデータの形式は、buffer

受信 cb へのインスタンス書き込みを実行し、clearBuffer を呼び出してキャッシュをクリアします
初めての読み取りかどうかを判断します。初めての読み取りは、直接書き込み、_write (実装予定) を呼び出します。
キャッシュキューテールオファーは現在、書き込みを待機しているデータを読み取ります。

 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.writing = false; 次に読み取られたファイルは、ターゲットファイルに直接書き込むことができます。
this.needDrain が再び期待を満たした場合、ファイルが読み取られ、データは直接ターゲットファイルに書き込まれずにキャッシュに保存されます。 target file

clearBuffer() {
    //写入成功后 调用 clearBuffer--》写入缓存第一个，第一个完成后，再继续 第二个
    let data = this.cache.poll();
    // console.log(&#39;this.cache&#39;,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 は数値型になります

fd タイプは数値です。fs.write を呼び出して現在のデータを書き込みます。チャンク、

 _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 での書き込み可能なストリームの書き込みと実装方法に関する簡単な説明期待した値がターゲットに正しく書き込まれているか確認してくださいファイル