NodeJS -- 实现一个可读流（流动模式）

灰太狼 2022-02-13 08:51 213阅读 0赞

上一节，我们实现了一个可写流[可写流][Link 1]

那么我们这一节实现一个可读流的（流动模式）。nodeJs有俩种模式的可读流，一种是暂停模式，一种是流动模式。

什么是流动模式了？监听data事件，调用resume()方法，使用Pipe()都是流动模式，流动模式就是说数据是不会经过缓冲区的，像水流一样，只要打开就不停的往下面流。暂停模式就像你用吸管吸一样，你像要多少数据，你就吸多少。

流动模式的实现比较简单，就是利用fs.read方法，每一次去读取数据，然后判断是否读取完了，否者就继续读。

const EventEmitter  = require("events");
    const fs = require("fs");
    
    
    class ReadStream extends EventEmitter {
        constructor(path, options) {
            super();
            this.path = path;
            this.flags = options.flags || "r";
            this.mode = options.mode || 0o666;
            this.highWaterMark = options.highWaterMark || 16 * 2024;
            this.encoding = options.encoding || "utf8";
            this.start = options.start || 0;
            this.end = options.end;
            this.autoClose = options.autoClose || true;
    
            this.fd;
            this.buffer = Buffer.alloc(this.highWaterMark);
            this.flowing = undefined; //表示当前处于何种工作模式
            this.pos= this.start || 0; //当前读入的位置
            this.open(); //打开文件夹
    
            this.on("newListener", (type, linstener) => {
                if (type == "data") { //流动模式
                    this.flowing = true; //流动模式
                    this.read();
                }
            })
        }
    
        open () {
            fs.open(this.path, this.flags, this.mode, (err, fd) => {
                if (err) {
                    if (this.autoClose) {
                        fs.close(fd, () => { this.emit("close", "异常关闭"); }); //关闭文件
                    }
                    this.emit("error", "异常出错");
                }
                this.fd = fd;
               // console.log(this.fd);
                this.emit("open", "文件打开了");
            })
        }
    
        read () {
            if (typeof this.fd !== "number") { //如果还未打开，那么就继续发送open事件
                return this.once("open", () => this.read()  );
            }
            //如果存在end那么就看最后一次是否能够读取this.highWaterMark的字节数
            let readBytes = this.end?Math.min(this.end-this.pos+1, this.highWaterMark):this.highWaterMark;
           // console.log(this.fd);
            fs.read(this.fd, this.buffer, 0, readBytes, this.pos, (err, bytes) => {
                if (err) {
                    if (this.autoClose) {
                        fs.close(this.fd, () => {this.emit("close", "异常关闭"); })
                    }
                    this.emit("error", "异常出错");
                }
                if (bytes) {
                    let data = this.buffer.slice(0, bytes);
                    data = this.encoding?data.toString(this.encoding):data; //如果有encoding，那么就进行转码
                    this.pos += bytes;//加上已经读入的字符数
                    //console.log(this.pos);
                    this.emit("data", data); //发送data事件
                    if (this.end&& this.end>this.pos) {
                        this.emit("end", "读取结束");
                        fs.close(this.fd, () => {this.emit("close", "正常关闭"); })
                    } else {
                        if (bytes && this.flowing) this.read(); //如果是流动模式 继续读
                    }
                } else {
                    this.emit("end", "读取结束")
                    fs.close(this.fd, () => {this.emit("close", "正常关闭"); })
                }
            })
        }
    
        pipe (dist) {
            this.on("data", (data) => {//开始读取数据
                let flag = dist.write(data);// 往里面写入
                if (!flag) {
                    this.pause();//如何存不下就停止
                }
    
            })
            dist.on("drain", () => {
                this.resume(); //恢复
            })
        }
        pause () {
            this.flowing = false; //开启暂停模式
        }
        resume () {
            this.flowing = true; //开启流动模式
        }
    }
    
    module.exports = ReadStream;

我们简单的测试一下

const ReadStream = require("./readStream");
    const WriteStream = require("./WriteSteam");
    const rs = new ReadStream("./1.txt", {
        start: 2,
        end: 4,
        highWaterMark: 3
    });
    const ws = new WriteStream("./2.txt", {
        highWaterMark: 3
    });
    
    rs.pipe(ws);

![20190427152108594.png][]![20190427152116656.png][]

[Link 1]: https://blog.csdn.net/wust_cyl/article/details/89506022
[20190427152108594.png]: https://img-blog.csdnimg.cn/20190427152108594.png
[20190427152116656.png]: /images/20220213/003dafc76daf48829c982e571ad7ff07.png

发表评论取消回复

表情：

评论列表（有 0 条评论，213人围观）

还没有评论，来说两句吧...

相关阅读

相关 js实现一个节流

节流函数通常用于限制函数在指定时间内只能调用一次，以节省资源。例如，在页面的滚动事件中，您可能希望仅在用户停止滚动一段时间后才执行函数，而不是每次滚动时都执行函数。下面是使

客官°小女子只卖身不卖艺/ 2024年03月25日 13:14/ 0 赞/ 11 阅读

相关 CSS实现流动边框

【写在前面】咳咳，首先讲个题外话，原本我是做 C++/Qt 的，然而因为工作 ( 部分原因 ) 学起了前端。当然，这也并不意味着我会丢掉 C++/Qt/Qml 这些

今天药忘吃喽~/ 2022年12月13日 11:27/ 0 赞/ 231 阅读

相关 MVCC如何实现提交读和可重复读？

MVCC（多版本并发控制）多版本并发控制（MultiVersion Concurrency Control，简称 MVCC 或 MCC），也称为多版本数据库：不用加任何

「爱情、让人受尽委屈。」/ 2022年11月29日 12:48/ 0 赞/ 116 阅读

相关 javaIO流实现读写txt文件

javaIO流实现文件读写文件写入: InputStreamReader BufferedReader 文件读取: File

╰半橙微兮°/ 2022年08月23日 14:44/ 0 赞/ 136 阅读

相关 NodeJS -- 简单实现一个可写流

实现node.js里面的可写流主要在于俩个点 1：有一个缓冲区，用于对没有来得及写入的数据进行缓存 2：有一个字段用来标记当前是否为写入状态。在代码体现为, this.

红太狼/ 2022年04月24日 11:54/ 0 赞/ 165 阅读

相关 NodeJS -- 实现一个可读流（流动模式）

上一节，我们实现了一个可写流[可写流][Link 1] 那么我们这一节实现一个可读流的（流动模式）。nodeJs有俩种模式的可读流，一种是暂停模式，一种是流动模式。什么是

灰太狼/ 2022年02月13日 08:51/ 0 赞/ 214 阅读

相关 nodejs stream流

今天起得晚，发现我的博客排名上升到5位数了，激动之情难以言表，只有再写一篇博客来巩固自己的地位（以上内容与本文无关，轻喷~）之前说过，nodejs操

系统管理员/ 2022年02月09日 13:27/ 0 赞/ 182 阅读

相关 NodeJS -- 简单的实现一个登录功能

实现目标：默认起始页面时登录页面，输入账号密码，如果没有此账号那么跳到注册页面，注册成功后进入个人页面。登录时，如果账号密码不匹配，那么跳到错误页面，三秒后跳到登录页面。

谁践踏了优雅/ 2022年01月23日 07:09/ 0 赞/ 324 阅读

相关流动比率

流动比率是流动资产对流动负债的比率，用来衡量企业流动资产在短期债务到期以前，可以变为现金用于偿还负债的能力。一般说来，比率越高，说明企业资产的变现能力越强，短期偿债能力亦越强；

你的名字/ 2021年07月05日 01:36/ 0 赞/ 371 阅读

相关 Node.js中Stream-可读流的使用

Node.js中Stream-可读流的使用可读流是生产数据 const rs = fs.createReadStream(filePath); rs 就是一个可读流，其...

红太狼/ 2020年06月22日 15:09/ 0 赞/ 695 阅读