Une brève discussion sur les méthodes d'écriture et d'implémentation de flux inscriptibles dans Nodejs-js tutoriel-php.cn

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.

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

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

Copier après la connexion

Effet

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

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;

Copier après la connexion

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");

Copier après la connexion

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

Copier après la connexion

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

Copier après la connexion

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

Copier après la connexion

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

Copier après la connexion

_write()

fs.open() est asynchrone Après une lecture réussie, fd sera un type numérique <🎜. >

 _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

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