文件流

我们需要读文件,有时候并不需要全部读出来,因为文件可能会很大,可能会有上百G,我们不可能把他全部拷贝到内存中,像前面的fs.readFile方法会将一整个文件全部读取到内存中,这显然不是我们需要的,不然内存直接爆炸了,我们希望一点一点的读文件,这就是文件流.

绪论

所谓的流,就是数据的流动,就是数据从一个地方慢慢流到另一个地方,既然是从一个地方流向另一个地方,可见,流一定是有方向的,在nodejs中,我们把流的方向分为三种,可读流,可写流,双工流

• 可读流：数据从源头流向内存就是可读流,这其实很好理解,所谓的可读流,就是读取文件的流
• 可写流：数据从内存流向源头就是可写流,也就是写入文件的流向
• 双工流：这个流可写可读

源头不一定是硬盘,也可以是网卡,显卡等其他设备

那么为什么需要流呢,主要总结为两个原因

1. **其他介质与内存的数据规模不一致
   **比如硬盘可能有几个T,但内存远达不到这个量级,内存能容纳的数据远不如磁盘,当我们读大量数据的时候,我们只能采取一点一点读的策略,不可能把所有内容全部读入内存.
2. 其他介质与内存的处理能力不一致
   内存的传输速度相当之快,而形如硬盘等外设的速度是非常慢的,如果内存的数据一股脑全给硬盘,显然速度是不匹配的,内存很快就可以传输完,但硬盘需要很长时间才能写入,会导致硬盘卡顿,我们可以以流的方式慢慢传给硬盘,让硬盘在保存这些数据的同时不至于过于卡顿.

Nodejs中提供了一个stream模块,这个模块里给我们提供了两个类,Readalbe和Writable类,所有的流都应该去继承这两个类,所以所有的流都应该有一些共同的属性和方法,当学完了文件流之后,其他的流的操作也是类似的.

所谓的文件流,就是内存和磁盘之间的数据流,数据从硬盘流入到内存就是可读流,从内存流入到硬盘就是可写流.

文件可读流

我们可以通过fs的模块的createReadStream方法创建一个文件可读流,第一个参数填入路径,第二个参数填入配置对象.当我们调用了这个方法后,文件会自动打开.

const rs = fs.createReadStream(fileName, {
    //创建一个文件可读流，此时会打开这个文件
    encoding: "utf-8",
    highWaterMark: 12,
});

常用的配置信息如下

• encoding：编码方式,如果不指定的话默认为null,读出来的就是buffer
• start：起始字节,我可以指定从第几个字节开始读,我们一般是全读出来,不会管.
• end：结束字节
• highWaterMark：每次读取的字节数,默认值是64*1024,也就是64kb的数量
• autoClose：文件读取完毕后是否会自动关闭,默认为true

调用这个方法后会返回一个继承自readable的readstream对象.

这个有一个on方法,可以给他绑定事件,具体事件及其作用参考以下代码.

const fs = require("fs");
const path = require("path");
const fileName = path.resolve(__dirname, "./test.txt");
const rs = fs.createReadStream(fileName, {
    //创建一个文件可读流，此时会打开这个文件
    encoding: "utf-8",
    highWaterMark: 6,
});
rs.on("open", () => {
    //文件打开事件，文件被打开的时候触发这个事件
    console.log("文件被打开");
});
rs.on("error", () => {
    //读取文件出错的时候触发该事件，比如文件不存在
});
rs.on("close", () => {
    //文件关闭事触发
    console.log("文件被关闭");
});
// rs.close();//我们可以调用close事件来手动关闭文件，文件读取完毕后会自动关闭（这取决于autoClose配置，我们一般是不会改变的）

rs.on("data", (chunk) => {
    //highWaterMark属性决定了我们每一次读取到多少数据，所以我们的数据实际上是分成很多次来读取的，每一次读取highWaterMark这么多的数据，每一次数据读取到之后，就会触发data事件，所以data事件会触发非常多次
    // 这个参数的回调函数会给我们传一个参数，这个参数就是本次读到的数据，我们可以在函数体里对数据进行操作
    //绑定了data函数后，就会立刻开始读取数据
    console.log(chunk); //我们可以输出测试一下我们得到的数据
});
rs.on("end", () => {
    //全部数据读取完毕后触发，和close有区别，如果同时有close和end事件，先触发end，后触发close
    console.log("end被触发");
});

很显然,这种方式所占用的内存就会很少

除此之外,rs对象还提供了两个常用的方法

rs.pause()方法可以暂停读取,rs.resume()可以恢复读取

rs.pause()被调用时会触发pause事件,恢复读取会触发resume事件,这两个事件也需要on方法绑定

const fs = require("fs");
const path = require("path");
const fileName = path.resolve(__dirname, "./test.txt");
const rs = fs.createReadStream(fileName, {
    //创建一个文件可读流，此时会打开这个文件
    encoding: "utf-8",
    highWaterMark: 6,
});
rs.on("data", (chunk) => {
    console.log(chunk);
    rs.pause(); //每次读取6个字节，然后暂停
});
rs.on("pause", () => {
    setTimeout(() => {
        rs.resume(); //等待半秒钟继续读取
    }, 500);
});

文件可写流

可写流的创建和可读流类似,第一个参数为路径,第二个参数为配置对象

const fs = require("fs");
const path = require("path");
const fileName = path.resolve(__dirname, "./test1.txt");
const ws = fs.createWriteStream(fileName, {
    //创建一个文件可读流，此时会打开这个文件
    encoding: "utf-8",
    highWaterMark: 12,
  	flags:"a",
});

常见配置如下

• flags：写入类型,a为追加,w为覆盖,默认为追加
• encoding：默认为UTF-8,如果设置为null,需要写入buffer
• start：起始字节
• highWaterMark：每次最多写入的字节数

返回一个WriteStream类的对象

写入流也有一些事件可以用on方法绑定,主要有三个事件,分别是open,error,close,这三个事件和读取流一样,不赘述

除此之外,还有一些自己独有的方法,写入文件的方法为ws.write(data),调用这个方法后可以实现写入,传入一个参数,这个参数可以是字符串,也可以是buffer,调用这个方法后就会将数据写入到路径中

ws.write()方法会返回一个布尔值,如果返回true,表示写入通道没有被塞满,接下来的数据可以写入,不用排队,如果返回false,表示目前通道已经被塞满了,接下来的数据将会进入写入队列.

磁盘和内存其实有一个缓存通道,这个通道能装入的字节数就是我们highWaterMark里面设置的,内存将数据传入通道,磁盘从通道拿取数据,这就是一个写入的流程,但是磁盘的速度远不如内存的速度,如果我们通道内存很小,又很快速地写入大量数据,通道就会被塞满,如果write返回true,表示通道没有被塞满,如果write返回false,表示通道已经被塞满了,下一次不能再直接写入到通道,需要在内存排队

如果内存里有数据在排队,说明通道已经被塞满了,但如果我们还继续写数据,就会全放在队列中,就会给队列很大的压力,这就是背压问题,由于这个队列是放在内存中的,所以背压问题会导致占用很多内存,我们一定要注意背压问题.

const fs = require("fs");
const path = require("path");
const fileName = path.resolve(__dirname, "./test1.txt");
const ws = fs.createWriteStream(fileName, {
    //创建一个文件可读流，此时会打开这个文件
    encoding: "utf-8",
    highWaterMark: 12,
});
for (let i = 0; i < 1024 * 1024 * 10; i++) {
    ws.write("a"); //我们写入10M个a
}

我们如何来解决背压问题呢?当队列清空的时候,会触发drain事件,我们可以利用drain事件和write的返回值,当write返回false时,我们需要等待一会儿,等他触发drain事件后再写入.我们改进上面的代码.

const fs = require("fs");
const path = require("path");
const fileName = path.resolve(__dirname, "./test1.txt");
const ws = fs.createWriteStream(fileName, {
    //创建一个文件可读流，此时会打开这个文件
    encoding: "utf-8",
    highWaterMark: 12,
});
let i = 0; //当前写入了多少
function write() {
    let flag = true; //下一次是否可以写入，最开始是可以写入的
    while (i < 1024 * 1024 * 10 && flag) {
        //我们写入10M，如果没有到达10M，并且还可以写入，我们执行循环
        flag = ws.write("a"); //修改flag
        i++; //修改i
    }
}
ws.on("drain", () => {
    //队列清空后，再调用write继续写
    write();
});
write();

写入流还提供了一个end方法,当我们调用end方法的时候,就会结束写入（如果autoClose为true,自动关闭文件）,里面有一个可选参数,表示最后一次的

学了可读流和可写流,我们就可以使用流操作,把一个文件读出来,复制到另一个文件

const fs = require("fs");
const path = require("path");
const to = path.resolve(__dirname, "./test2.txt");
const from = path.resolve(__dirname, "./test1.txt");
const rs = fs.createReadStream(from, {
    highWaterMark: 128, //每次读取128字节
});
const ws = fs.createWriteStream(to, {
    highWaterMark: 128,
});

rs.on("data", (chunk) => {
    //每次读取数据时，将数据写入
    if (!ws.write(chunk)) {
        //将数据写入,并且判断是否可以再次写入
        //如果产生了背压，暂停读取
        rs.pause();
    }
});
ws.on("drain", () => {
    //产生背压后，如果队列清空，继续读取
    rs.resume();
});

当然,我们也可以用文件IO直接实现

除了这种方式外,我们还可以使用管道实现,读取流里面有一个pipe方法,传入一个写入流,这个方式就可以把一个读取流和一个写入流接起来,当这两个流接起来后,他会自动将读取流的文件复制到写入流,这个方法自动处理背压问题