Une brève discussion sur les méthodes d'écriture et d'implémentation de flux inscriptibles dans Nodejs

Cet article vous guidera à travers l'écriture de flux inscriptible dans Nodejs et présentera l'implémentation de l'écriture de flux inscriptible de Node. Il a une certaine valeur de référence. Les amis dans le besoin peuvent s'y référer. J'espère qu'il sera utile à tout le monde.

[Apprentissage recommandé : "Tutoriel Nodejs"]

Flux inscriptible-Inscriptible

Exemple d'appel fs.createWriteStream

• Les données lues pour la première fois seront en fait écrites dans le fichier cible
• Les données lues pour les fois restantes sera lu selon si les données dépassent highWaterMark, si tel est le cas, elles sont stockées dans la zone tampon et attendent d'être écrites dans le fichier cible

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是否有打印

• Effet

Flux initWriteStream personnalisé en écriture

Hérite de l'abonnement de publication EventEmitter

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

Lié File d'attente de génération de liste pour la mise en cache de lecture de fichiers

Mise en œuvre de la liste chaînée et de la file d'attente

https://juejin.cn/post/6973847774752145445

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

Initialiser le constructeur de données par défaut de l'instance()

 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();
  }

• L'autorisation d'opération de fichier.mode est par défaut 0o666 (0o signifie octal)

  • Les positions occupées par les trois 6 correspondent respectivement : aux permissions de l'utilisateur auquel appartient le fichier ; aux permissions du groupe d'utilisateurs auquel appartient le fichier indiquant les permissions des autres utilisateurs sur celui-ci

  • Les autorisations sont : r--lisible (valeur correspondante 4), w--inscriptible (correspondant à la valeur 2), x--exécutable (correspondant à la valeur 1, par exemple, s'il y a un Marque .exe sous le dossier, cela signifie que le clic peut être exécuté directement) pour former

  • Donc par défaut, les trois groupes d'utilisateurs ont des autorisations de lecture et d'écriture sur les fichiers

open()

• Appelez fs.open()
• le rappel émet la méthode d'ouverture de l'instance, La valeur de retour de fs.open fd est passée en paramètre

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

write ()

• Le format des données du fichier qui doivent être écrites l'instance de conversion est buffer
• Déterminez si la longueur des données écrites est supérieure à highWaterMark. Si elle atteint l'attente, le fichier lu Les données obtenues sont stockées dans le cache et ne sont pas directement écrites dans le fichier cible. (cela exclut si c'est la première fois que vous lisez le fichier)
• Exécutez l'écriture d'instance dans le cb transmis et appelez clearBuffer pour vider le cache
• Déterminez s'il s'agit de la première lecture. La première lecture écrit et appelle directement _write (à implémenter)
• L'offre en fin de file d'attente du cache lit actuellement les données en attente d'écriture dans le fichier cible

 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() efface la file d'attente du cache dans l'ordre

• Ordre d'exécution de la file d'attente, principe du premier entré, premier sorti
• this.cache.poll() dans séquence Récupérez les données d'en-tête et exécutez this._write pour écrire dans le fichier cible
• Si les données interrogées par la file d'attente du cache n'existent pas, cela signifie que c'est le premier comportement d'écriture || La file d'attente du cache a été vide. this.writing = false; Le prochain fichier lu peut être écrit directement dans le fichier cible
• Si this.needDrain répond à nouveau aux attentes, le fichier est lu et les données sont stockées dans le cache sans être directement écrites dans le fichier cible

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() est asynchrone Après une lecture réussie, fd sera un type numérique <🎜. >
Décidez si vous souhaitez vous abonner à un open en fonction du type de fd et rappelez-vous (jusqu'à ce que le type de fd soit un numéro)
le type de fd est un numéro : appelez fs.write, écrivez le morceau actuel ,

 _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);
    });
  }

Copier après la connexion

Testez le personnalisé 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();

Copier après la connexion

10 nombres, écrits dans l'ordre, atteignant la valeur maximale attendue 3 fois, puis effaçant le cache 3 fois de suite, le résultat est conforme aux attentes

Vérifiez si la valeur attendue est correctement écrite dans le fichier cible

Pour plus de connaissances liées à la programmation, veuillez visiter :

Vidéo de programmation ! !

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!