当前位置: 首页 > 编程笔记 >

浅谈手写node可读流之流动模式

甘祺
2023-03-14
本文向大家介绍浅谈手写node可读流之流动模式,包括了浅谈手写node可读流之流动模式的使用技巧和注意事项,需要的朋友参考一下

node的可读流基于事件

可读流之流动模式,这种流动模式会有一个"开关",每次当"开关"开启的时候,流动模式起作用,如果将这个"开关"设置成暂停的话,那么,这个可读流将不会去读取文件,直到将这个"开关"重新置为流动。

读取文件流程

读取文件内容的流程,主要为:

  1. 打开文件,打开文件成功,将触发open事件,如果打开失败,触发error事件和close事件,将文件关闭。
  2. 开始读取文件中的内容,监听data事件,数据处于流动状态,可通过修改开关的状态来暂停读取。
  3. 每次读取到的内容放入缓存中,并通过data事件将数据发布出去。
  4. 当文件中的内容读取完毕之后,将文件关闭。

这一系列动作都是基于事件来进行操作的,而node中的事件我们都知道是一种发布订阅模式来实现的。

下面我们来看一看,node是如何使用可读流来读取文件中的内容?

node 可读流参数

首先我们通过fs模块来创建一个可读流,可读流接受两个参数:

  1. 第一个参数是要读取的文件地址,在这里指明你要读取哪个文件。
  2. 第二个参数是可选项,这个参数是一个对象,用来指定可读流的一些具体的参数。

如下几个参数我们来一一说明:

  • highWaterMark:设置高水位线,这个参数主要用于在读取文件时,可读流会将文件中的内容读取到缓存当中,而这里我们需要创建一个buffer来缓存这些数据,所以这个参数是用来设置buffer的大小,如果不对这个参数进行设置的话,可读流默认的配置64k。
  • flags:这个参数主要用于设置文件的执行模式,比如说我们具体的操作适用于读取文件还是写入文件等这些操作。如果是写入文件的话那我们,使用的是w。如果是读取文件的话那这个操作符就应该是r。

下面这张表格就说明了不同的符号代表不同含义:

符号 含义
r 读文件,文件不存在报错
r+ 读取并写入,文件不存在报错
rs 同步读取文件并忽略缓存
w 写入文件,不存在则创建,存在则清空
wx 排它写入文件
w+ 读取并写入文件,不存在则创建,存在则清空
wx+ 和w+类似,排他方式打开
a 追加写入
ax 与a类似,排他方式写入
a+ 读取并追加写入,不存在则创建
ax+ 作用与a+类似,但是以排他方式打开文件

  • autoClose:这个参数主要用于,对文件的关闭的一些控制。如果文件再打开的过程或者其他操作的过程中出现了错误的情况下,需要将文件进行关闭。那这个参数是设置文件是否自动关闭的功能。
  • encoding:node中用buffer来读取文件操作的东西二进制数据。这些数据展现出来的话我们是一堆乱码,所以需要,要我们对这个数据指定一个具体的编码格式。然后将会对这些数据进行编码转化,这样转化出来的数据就是我们能看懂的数据
  • starts:这个参数主要用于指定从什么位置开始读取文件中的内容,默认的话是从零开始。
  • ends:这个参数主要用于指定定具体要读取文件多长的数据,这里需要说明一下,这个参数是包括本身的位置,也就是所谓的包前和包后。

下面我们来看看可读流具体例子:

let fs = require("fs");
let rs = fs.createReadStream("./a.js", {
  highWaterMark: 3,
  encoding: "utf8",
  autoClose: true,
  start: 0,
  end: 9
});
rs.on("open", () => {console.log("open");});
rs.on("close", () => {console.log("close");});
rs.on("data", data => {
  console.log(data);
  rs.pause();//暂停读取 此时流动模式为暂停模式
});
setInterval(() => {
  rs.resume();//重新设置为流动模式,开始读取数据
}, 1000);
rs.on("end", () => { console.log("end"); });
rs.on("error", err => { console.log(err); });

手写可读流第一步

上面我们说过,node可读流是基于node的核心模块事件来完成的,所以在实现我们自己的可读流时需要继承events模块,代码如下:

let fs = require('fs');
let EventEmitter = require('events');
class ReadStream extends EventEmitter {

}

继承了EventEmitter类,我们就可以使用EventEmitter类中的各个方法,并且同样是采用发布订阅的模式了处理事件。

第二步:处理可读流配置的参数

上面我们提到,node中创建可读流时可以对这个流配置具体的参数,比如

let rs = fs.createReadStream("./a.js", {
  highWaterMark: 3,
  encoding: "utf8",
  autoClose: true,
  start: 0,
  end: 9
});

那么对于这些参数,我们自己实现的可读流类也需要对这些参数进行处理,那么这些参数该如何进行处理呢?

constructor(path, options = {}) {
  super();
  this.path = path; //指定要读取的文件地址
  this.highWaterMark = options.highWaterMark || 64 * 1024;
  this.autoClose = options.autoClose || true; //是否自动关闭文件
  this.start = options.start || 0; //从文件哪个位置开始读取
  this.pos = this.start; // pos会随着读取的位置改变
  this.end = options.end || null; // null表示没传递
  this.encoding = options.encoding || null;// buffer编码
  this.flags = options.flags || 'r';

  this.flowing = null; // 模式开关
  this.buffer = Buffer.alloc(this.highWaterMark);// 根据设置创建一个buffer存储读出来的数
  this.open();
}

通常配置的原则是以用户配置的参数为准,如果用户没有对这个参数进行设置的话,就采用默认的配置。

实现可读流第三步:打开文件

这里原理是使用node模块fs中的open方法。首先我们来回顾下fs.open()方法的使用。

fs.open(filename,flags,[mode],callback);
//实例
fs.open('./1,txt','r',function(err,fd){});

这里需要说明下,回调函数callback中有2个参数:

  1. 第一个是error,node中异步回调都会返回的一个参数,用来说明具体的错误信息
  2. 第二个参数是fd,是文件描述符,用来标识文件,等价于open函数的第一个参数

好了,现在我们来看看我们自己的可读流的open方法该如何实现吧:

open() {
  fs.open(this.path, this.flags, (err, fd) => { 
    //fd标识的就是当前this.path这个文件,从3开始(number类型)
    if (err) {
      if (this.autoClose) { // 如果需要自动关闭则去关闭文件
        this.destroy(); // 销毁(关闭文件,触发关闭事件)
      }
      this.emit('error', err); // 如果有错误触发error事件
      return;
    }
    this.fd = fd; // 保存文件描述符
    this.emit('open', this.fd); // 触发文件的打开的方法
  });
}

从代码上我们可以看出:

fs.open函数是异步函数,也就是说callback是异步执行的,在成功打开文件的情况下,fd这个属性也是异步获取到的,这点需要注意。

另外重要的一点是,如果在打开文件发生错误时,则表明打开文件失败,那么此时就需要将文件关闭。

实现可读流第四步:读取文件内容

上面我们详细说过,可读流自身定义了一个"开关",当我们要读取文件中的内容的时候,我们需要将这个"开关"打开,那么node可读流本身是如何来打开这个"开关"的呢?

监听data事件

node可读流通过监听data事件来实现这个"开关"的开启:

rs.on("data", data => {
  console.log(data);
});

当用户监听data事件的时候,"开关"开启,不停的从文件中读取内容。那么node是怎么监听data事件的呢?
答案就是 事件模块的newListener

这是因为node可读流是基于事件的,而事件中,服务器就可以通过newListener事件监听到从用户这边过来的所有事件,每个事件都有对应的类型,当用户监听的是data事件的时候,我们就可以获取到,然后就可以去读取文件中的内容了,那我们自己的可读流该如何实现呢?

// 监听newListener事件,看是否监听了data事件,如果监听了data事件的话,就开始启动流动模式,读取文件中的内容
this.on("newListener", type => {
  if (type === "data") {
    // 开启流动模式,开始读取文件中的内容
    this.flowing = true;
    this.read();
  }
});

好了,知道了这个"开关"是如何打开的,那么这个时候就到了真正读取文件中内容的关键时候了,先上代码先:

read() {
  // 第一次读取文件的话,有可能文件是还没有打开的,此时this.fd可能还没有值
  if (typeof this.fd !== "number") {
    // 如果此时文件还是没有打开的话,就触发一次open事件,这样文件就真的打开了,然后再读取
    return this.once("open", () => this.read());
  }
  // 具体每次读取多少个字符,需要进行计算,因为最后一次读取倒的可能比highWaterMark小
  let howMuchRead = this.end ? Math.min(this.end - this.pos + 1, this.highWaterMark) : this.highWaterMark;
  fs.read(this.fd, this.buffer, 0, howMuchRead, this.pos, (err, byteRead) => {
    // this.pos 是每次读取文件读取的位置,是一个偏移量,每次读取会发生变化
    this.pos += byteRead;
    // 将读取到的内容转换成字符串串,然后通过data事件,将内容发布出去
    let srr = this.encoding ? this.buffer.slice(0, byteRead).toString(this.encoding) : this.buffer.slice(0, byteRead);
    // 将内容通过data事件发布出去
    this.emit("data", srr);
    // 当读取到到内容长度和设置的highWaterMark一致的话,并且还是流动模式的话,就继续读取
    if ((byteRead === this.highWaterMark) && this.flowing) {
      return this.read();
    }
    // 没有更多的内容了,此时表示文件中的内容已经读取完毕
    if (byteRead < this.highWaterMark) {
      // 读取完成,发布end方法,并关闭文件
      this.emit("end");
      this.destory();
    }
  });
}

这里我们特别要注意的是:

  1. 文件是否已经打开,是否获取到fd,如果没有打开的话,则再次触发open方法
  2. 分批次读取文件内容,每次读取的内容是变化的,所以位置和偏移量是要动态计算的
  3. 控制读取停止的条件。

实现可读流第五步:关闭文件

好了,到现在,基础的读取工作已经完成,那么就需要将文件关闭了,上面的open和read方法里面都调用了一个方法:destory,没错,这个就是关闭文件的方法,好了,那么我们来看看这个方法该如何实现吧

destory() {
  if (typeof this.fd !== "number") {
    // 发布close事件
    return this.emit("close");
  }
  // 将文件关闭,发布close事件
  fs.close(this.fd, () => {
    this.emit("close");
  });
}

当然这块的原理就是调用fs模块的close方法啦。

实现可读流第六步:暂停和恢复

既然都说了,node可读流有一个神奇的"开关",就像大坝的阀门一样,可以控制水的流动,同样也可以控制水的暂停啦。当然在node可读流中的暂停是停止对文件的读取,恢复就是将开关打开,继续读取文件内容,那么这两个分别对应的方法就是pause()和resume()方法。

那么我们自己的可读流类里面该如何实现这两个方法的功能呢?非常简单:

我们在定义类的私有属性的时候,定义了这样一个属性flowing,当它的值为true时表示开关打开,反之关闭。

pause() {
  this.flowing = false;// 将流动模式设置成暂停模式,不会读取文件
}
resume() {
  this.flowing = true;//将模式设置成流动模式,可以读取文件
  this.read();// 重新开始读取文件
}

好了,关于node可读流的实现我们就写到这里,快快敲起代码,动手实现一个你自己的可读流吧!

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持小牛知识库。

 类似资料:
  • 本文向大家介绍说说node中的可读流和可写流的区别,包括了说说node中的可读流和可写流的区别的使用技巧和注意事项,需要的朋友参考一下 前言 nodejs中大量的api与流有关,曾经看到公司的一些大神的node代码,实现一个接口只需要pipe一下另一个java接口就可以了。简单的一行代码实在让人困惑。作为小白的自己一脸懵逼却又不敢问,因为根本不知道从何问起。现在终于通过学习,也能对流说出个123,

  • 本文向大家介绍浅谈JavaScript函数节流,包括了浅谈JavaScript函数节流的使用技巧和注意事项,需要的朋友参考一下 浏览器中某些计算和处理要比其他的昂贵的多。例如,DOM操作比起非DOM交互需要更多的内存和CPU时间。连续尝试进行过多的DOM相关操作可能会导致 浏览器挂起,有时候甚至会崩溃。尤其在IE中使用onresize事件处理程序的时候容易发生,当调整浏览器大小的时候,该事件连续触

  • 本文向大家介绍浅谈JS函数节流防抖,包括了浅谈JS函数节流防抖的使用技巧和注意事项,需要的朋友参考一下 在前端开发中有一部分的用户行为会频繁的触发事件执行,而对于DOM操作、资源加载等耗费性能的处理,很可能导致界面卡顿,甚至浏览器的崩溃。函数节流(throttle)和函数防抖(debounce)就是为了解决类似需求应运而生的。 函数节流(throttle) 函数节流就是预定一个函数只有在大于等于执

  • 再说启动流程之前我们看看一个github的项目。 cobra 它是一个是用于创建强大的现代CLI应用程序的库,也是用于生成应用程序和命令文件的程序。 简单点来说就是方便使用者更易创建命令行工具。 举个静态博客生成器hugo的例子: hugo help hugo is the main command, used to build your Hugo site. Hugo is a Fast an

  • 本文向大家介绍浅谈C#设计模式之代理模式,包括了浅谈C#设计模式之代理模式的使用技巧和注意事项,需要的朋友参考一下 代理模式是常用的结构型设计模式之一,当无法直接访问某个对象或访问某个对象存在困难时可以通过一个代理对象来间接访问,为了保证客户端使用的透明性,所访问的真实对象与代理对象需要实现相同的接口.根据代理模式的使用目的不同,代理模式又可以分为多种类型,例如保护代理、远程代理、虚拟代理、缓冲代

  • 本文向大家介绍浅谈JAVA设计模式之享元模式,包括了浅谈JAVA设计模式之享元模式的使用技巧和注意事项,需要的朋友参考一下 享元模式(Flyweight Pattern)主要用于减少创建对象的数量,以减少内存占用和提高性能。这种类型的设计模式属于结构型模式,它提供了减少对象数量从而改善应用所需的对象结构的方式。 享元模式尝试重用现有的同类对象,如果未找到匹配的对象,则创建新对象。我们将通过创建 5