一文带你搞懂Node中的流【nodejs教程】

目录

• 流是什么？
• 为什么需要流
• 文件读写
• 分片读写
• 可读流及源码编写
• EventListener
• Readable 可读流
• 源码实现
• 总结

const fs = require("fs");
const path = require("path");
// 利用 path 上的方法组装路径
fs.readFile(path.resolve(__dirname, "test.txt"), (err, data) => {
  if (err) return console.log("error", err);
  fs.writeFile(path.resolve(__dirname, "result.txt"), data, () => {
    console.log("拷贝成功");
  });
});

// 创建一个存储单位 1 的 Buffer 空间，来存储中间读取的数据
let buf = Buffer.alloc(1);
// 读取源文件中的数据
fs.open(path.resolve(__dirname, "test.js"), "r", function (err, rfd) {
  // rfd 可以理解为文字指针
  // 你看到了吗？6 个参数，麻爪
  // 甚至都有点解释不动
  fs.read(rfd, buf, 0, 1, 0, function (err, bytesRead) {
    // bytesRead读取到的字节长度
    // 读取到的第一个数据存入 buf 中
    console.log(buf); // 
    // 打开目标文件。
    fs.open(path.resolve(__dirname, "result.js"), "w", function (err, wfd) {
      // 6 个参数
      // 这里做的就是将 buf 内容写入 result
      fs.write(wfd, buf, 0, 1, 0, function (err, bytesWritten) {
        console.log("拷贝成功");
      });
    });
  });
});

// source 源文件
// target 目标文件
// cb 回调函数
// bufferSize buffer固定长度，即一次读写的数量
function copy(source, target, cb, bufferSize = 3) {
  const SOURCE_PATH = path.resolve(__dirname, source);
  const TARGET_PATH = path.resolve(__dirname, target);
  let buf = Buffer.alloc(bufferSize); // 创建 buffer 实例
  let rOffset = 0; // 读取偏移量
  let wOffset = 0; // 写入偏移量
  fs.open(SOURCE_PATH, "r", function (err, rfd) {
    if (err) return cb(err);
    fs.open(TARGET_PATH, "w", function (err, wfd) {
      if (err) return cb(err);
      // 递归读写函数 next
      function next() {
        fs.read(rfd, buf, 0, bufferSize, rOffset, function (err, bytesRead) {
          if (err) return cb(err);
          // bytesRead 代表一次读取的字节数
          // 当 bytesRead 为 0 时，代表文件已经成功读完
          // 则可以停止读写操作，关闭文件
          if (bytesRead == 0) {
            let index = 0;
            let dOne= () => {
              if (++index == 2) {
                cb();
              }
            };
            fs.close(wfd, done);
            fs.close(rfd, done);
            return;
          }
          fs.write(
            wfd,
            buf,
            0,
            bytesRead,
            wOffset,
            function (err, bytesWritten) {
              if (err) return cb(err);
              // 读取成功，并更新偏移量
              rOffset += bytesRead;
              wOffset += bytesWritten;
              next();
            }
          );
        });
      }
      next();
    });
  });
}
copy("test.js", "result.js", function (err) {
  if (err) return console.log(err);
  console.log("拷贝成功");
});

// 流是基于发布订阅模式实现的
// 因此我们只需要订阅对应事件即可
const fs = require("fs");
const path = require("path");
// 返回一个可读流
const rs = fs.createReadStream(path.resolve(__dirname, "test.txt"), {
  highWaterMark: 3, // 每次读取 3kb
});
// 文件打开的钩子函数
rs.on("open", (fd) => {
  console.log(fd); // 3
});
// 当可读流处于流动模式时，data 事件会不断触发
// 在这里我们可以获取到读取的数据，进行后续操作
rs.on("data", (chunk) => {
  console.log(chunk);
});
rs.on("end", () => {
  console.log("end"); // 结束事件
});

// 借助 pause 和 resume 方法可以实现数据读取的暂停与恢复
rs.on("data", function (data) {
  // 读取的数据为 buffer 类型
  console.log(`读取了 ${data.length} 字节数据 : ${data.toString()}`);

  //使流动模式的流停止触发"data"事件，切换出流动模式，数据都会保留在内部缓存中。
  rs.pause();

  //等待3秒后，再恢复触发"data"事件，将流切换回流动模式。
  setTimeout(function () {
    rs.resume();
  }, 3000);
});

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

constructor(path, optiOns= {}) {
  // 使用继承，子类必须调用 super 函数
  super();
  this.path = path; //指定要读取的文件地址
  this.highWaterMark = options.highWaterMark || 64 * 1024;
  this.autoClose = options.autoClose || true; //是否自动关闭文件
  this.start = options.start || 0; // 从文件哪个位置开始读取
  this.end = options.end || null; // null表示没传递
  this.encoding = options.encoding || null;// buffer编码
  this.flags = options.flags || "r";
}

constructor() {
  // ...
  this.pos = this.start;
  this.flowing = null;
  this.buffer = Buffer.alloc(this.highWaterMark);
}

open() {
    fs.open(this.path, this.flags, (err, fd) => {
        if (err) {
            if (this.autoClose) { // 如果需要自动关闭则去关闭文件
                this.destroy(); // 销毁(关闭文件，触发关闭事件)
            }
            this.emit("error", err); // 打开错误，触发 error 事件
            return;
        }
        this.fd = fd; // 保存文件描述符，方便后续轮询判断
        this.emit("open", this.fd); // 文件打开，触发 open 事件
    });
}

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

class ReadStream extends EventEmitter {
  constructor(path, options) {
    // 监听newListener事件，判断当前监听事件是否为 data 事件
    // 如果为 data 事件，开启文件读取
    this.on("newListener", (type) => {
      if (type === "data") {
        //  开启流动模式，开始读取文件中的内容
        this.flowing = true;
        this.read();
      }
    });
  }
}

read() {
    // 文件如果未打卡，fd 是没有值的
    if (typeof this.fd !== "number") {
        // 如果文件未打开，触发 open 事件
        return this.once("open", () => this.read());
    }
}

class ReadStream extends EventEmitter {
  read() {
    // 计算当前读取字节
    const howManyToRead = this.end
      ? Math.min(this.highWaterMark, this.end - this.pos + 1)
      : this.highWaterMark;
    // 创建 buffer 实例
    const buffer = Buffer.alloc(howManyToRead);
    // 利用 fs.read 进行文件内容读取
    fs.read(
      this.fd,
      buffer,
      0,
      howManyToRead,
      this.offset,
      (err, bytesRead) => {
        if (err) return this.destory(err);
        this.pos += bytesRead;
        // 可能存在最后一次的 buffer 大小 大于 实际数据大小的情况，所以使用slice来进行截取
        // 将读取后的内容传递给 data 事件
        this.emit("data", buffer.slice(0, bytesRead));
      }
    );
  }
}

read() {
    // ...
    // 流动模式下，循环进行读取
    if (this.flowing) {
         this.read();
    }
}

 pause() {
      // 判断当前是否读取完毕了
      if (this.flowing) {
        this.flowing = false;
      }
    }
    resume() {
      // 判断当前是否读取完毕了
      if (!this.flowing) {
        this.flowing = true;
        this.read();
      }
    }