本篇文章给大家介绍一下nodejs中的流（stream），看看node可读流的实现方法。有一定的参考价值，有需要的朋友可以参考一下，希望对大家有所帮助。
stream的概念流（stream）是 node.js 中处理流式数据的抽象接口。 stream 模块用于构建实现了流接口的对象。【推荐学习：《nodejs 教程》】
stream的作用读写大文件的过程中，不会一次性的读写到内存中。可以控制每次读写的个数
stream的分类1、可读流-readable
例：fs.createreadstream；
源码位置：lib/_stream_readable.js
2、可写流-writable
例：fs.createwritestream；
源码位置：lib/_stream_writable.js
3、双工流-duplex:满足读写的功能
例：net.socket()；
源码位置：lib/_stream_duplex.js
4、转化流-transform：用途：压缩，转码
例：
const { transform } = require('stream');transform.call(this, '要转换的数据');//具体的使用详情 见node官网
-源码位置：lib/_stream_tranform.js
可读流读取文件的过程读取文件代码过程const path = require("path");const apath = path.join(__dirname, "a.txt");//需要读取的文件const fs = require("fs");let rs = fs.createreadstream(apath, { flags: "r", encoding: null,//默认编码格式是buffer，深挖buffer又要学习字符编码，留个坑 到时候写一个编码规范的学习整理 autoclose: true,//相当于需要调用close方法，如果为false 文件读取end的时候 就不会执行 close start: 0, highwatermark: 3,//每次读取的个数 默认是64*1024个字节});rs.on("open", function (fd) { // fd number类型 console.log("fd", fd);});// 他会监听用户，绑定了data事件，就会触发对应的回调，不停的触发rs.on("data", function (chunk) {//这里会打印的是ascii 值 ，所以可以tostring自己看得懂的样子 console.log({ chunk }, "chunk.tostring", chunk.tostring()); //如果想每一段事件 读一点 可以用rs.pause() 做暂停,然后计时器 里rs.resume()再次触发data事件 rs.pause();//暂停读取});rs.on("close", function () { //当文件读取完毕后 会 触发 end事件 console.log("close");});setinterval(() => { rs.resume(); //再次触发data，直到读完数据为止}, 1000);
题外话：想说下 文件流和普通可读流的区别1、open 和close是文件流独有，支持open和close便是文件流
2、可读流都具备 （on('data'),on('end'),on('error'),resume,pause；所以只要支持这些方法就是可读流
可写流写入文件的过程写入文件代码过程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");});//write的参数string 或者buffer，ws.write 还有一个boolea的返回值表示是真实写入文件还是放入缓存中ws.write("1");let flag = ws.write("1");console.log({ flag });//trueflag = ws.write("1");console.log({ flag });//trueflag = ws.write("1");console.log({ flag });//false
双工流的写入和读取过程写一个本地服务 做例子1、server（服务器代码）实现
const net = require("net"); //net 模块是 node自己封装的tcp层//socket 就是双工流 能读能写 http源码就是用net模块写的 基于tcpconst server = net.createserver(function (socket) { socket.on("data", function (data) {//监听客户端发来的消息 console.log(data.tostring) socket.write("server:hello");//写入server：hello }); socket.on("end", function () { console.log("客户端关闭"); });});server.on("err", function (err) { console.log(err);});server.listen(8080);//服务端监听8080端口
2、client（客户端） 实现
const net = require("net"); //net 模块是 node自己封装的tcp层const socket = new net.socket(); //socket.connect(8080, "localhost"); // 表示链接服务器本地8080端口socket.on("connect", function (data) { //和服务器建立链接后 socket.write("connect server");});socket.on("data", function (data) { //监听数据，读取服务器传来的数据 console.log(data.tostring()); socket.destroy()});socket.write('ok')socket.on("error", function (err) { console.log(err);});
3.题外话 如果想看tcp的三次握手和四次挥手 可以 通过我上述代码 用wireshark（一个抓包工具）看实际过程
转化流 transform过程转化流是双工流的一种， 允许实现输入，并在对数据执行某些操作后返回输出，两者有依赖关系
代码过程(这个例子我的参考来处）const stream = require('stream')let c = 0;const readable = stream.readable({ highwatermark: 2, read: function () { let data = c < 26 ? number(c++ + 97) : null; console.log('push', data); this.push( string.fromcharcode(data));}})const transform = stream.transform({ highwatermark: 2, transform: function (buf, enc, next) { console.log('transform', buf.tostring()); next(null, buf); }})readable.pipe(transform);
打印结果
可读流的实现跟着断点先了解 可读流的调用过程
就前面可读流文件的读取过程的代码为例子 打断点
rs.on('open')rs.on('open')为断点入口进入
1、通过stream.prototype.on.call 继承stream类
源文件位置：no dlib/_stream_readable.js（我是通过断点点到这里 直接找，我也没找到）
再点进去 发现 stream 是eventemitter的子类 那么 可读流也可以支持发布订阅
2、监听的事件类型是否是data和readable任意一个 不是 继续 下一个事件的监听
rs.on('data')
data的部分做两件事
1、判断flowing（默认值是null）不为false 就自动resume方法执行继续 文件读取（这里我的案例是rs.pause();手动将flowing 值为false了所以不会继续调用）
2、那如果我没有调用rs.pause() 会继续调用resume 看看resume里做了什么
2.1 最终调用了 stream.read()继续读取文件;直到文件读取结束依次去emit end 和close事件
小结：所以data默认是会不断的读取文件直到文件读取完毕 ，如果想要文件读取变可控可以和我一样用rs.pause()
自己实现
实现思路
继承eventemitter发布订阅管理我们的事件const fs = require("fs");const eventemitter = require("events");class readstream extends eventemitter {}module.exports = readstream;
数据初始化constructor(path, options = {}) { super(); //参考fs 写实例需要用到的参数 this.path = path; this.flags = options.flags || "r"; this.encoding - options.encoding || null;//默认编码格式是buffer this.autoclose = options.autoclose || true;//相当于需要调用close方法，如果为false 文件读取end的时候 就不会执行 close this.start = options.start || 0;//数据读取的开始位置 this.end = options.end; this.highwatermark = options.highwatermark || 64 * 1024;//默认一次读取64个字节的数据 this.offset = this.start;//fs.read的偏移量 this.fd = undefined; //初始化fd 用于 open成功后的fd做赋值 供 read里使用 this.flowing = false;//实现pause和resume备用，设置flag,当监听到data事件的时候 改 flowing为true， this.open(); //初始化的时候就要调用open this.on("readstreamlistener", function (type) { // console.log(type)//这里打印就能看到 实例上所有 通过on 绑定的事件名称 if (type === "data") { //监听到data事件的时候 改 flowing为true this.flowing = true; this.read(); } }); }
文件读取方法read，pause，resume，open和destroy的实现open() open() { // 调用fs.open 读取目标文件 fs.open(this.path, this.flags, (err, fd) => { this.fd = fd; //赋值一个fd 供后面的 read()方式使用，文件读取成功，fd是返回一个数字 this.emit("open", fd); });
read() read() { // console.log("一开始read里的", this.fd); //但是这样依旧拿不到 open后的fd，用 发布订阅 通过on来获取 绑定的事件type //这里要做一个容错处理 ，因为open是异步读取文件，read里无法马上拿到open结果 if (typeof this.fd !== "number") { //订阅open，给绑定一个回调事件read 直到this.fd有值 return this.once("open", () => this.read()); } } //fd打开后 调用fs.read //实例上的start值是未知number,存在实际剩余的可读的文件大小<highwatermar的情况 ,用howmuchtoread 替换highwatermark 去做fs.read的每次读取buffer的大小 let howmuchtoread = this.end ? math.min(this.end - this.offset + 1, this.highwatermark) : this.highwatermark; //定义一个用户 传进来的highwatermark 大小的buffer对象 const buffer = buffer.alloc(this.highwatermark); //读取文件中的内容fd给buffer 从0位置开始，每次读取howmuchtoread个。插入数据,同时更新偏移量 fs.read( this.fd, buffer, 0, howmuchtoread, this.offset, (err, bytesread) => { if (bytesread) { // 每读完一次，偏移量=已经读到的数量 this.offset += bytesread; this.emit("data", buffer.slice(0, bytesread)); //写到这里实例上的data 已经可以打印出数据了 但是 继续读取 调用this.read() 直到bytesread不存在 说明数据读取完毕了 走else //回调 this.read();时候判断 this.flowing 是否为true //pause调用后this.flowing将为false if (this.flowing) { this.read(); } } else { // 执行到这 bytesread不存在说明 文件数据读取完毕了已经 触发end this.emit("end");//emit 实例上绑定的end事件 //destroy 还没写到 稍等 马上后面就实现... this.destroy(); } } );
resume()文件读取不去data事件，会触发对应的回调，不停的触发 所以想要变可控可以手动调用 resume（）& pause（）
pause的实现,调用的时候设置 this.flowing=false，打断 read（） pause() { this.flowing = false; }
pause()pause 打断 read（）多次读取，可以使用resume 打开 this.flowing=true 并调用readresume() { if (!this.flowing) { this.flowing = true; this.read(); } }
destroy()文件open不成功时候抛错时调用文件读取完毕后&&this.autoclose===true ，read()里文件读取end的时候 就执行close destroy(err) { if (err) { this.emit("error"); } // 把close放destroy里 并 在read里调用 if (this.autoclose) { fs.close(this.fd, () => { this.emit("close"); }); } }
完整代码实现代码/** *实现简单的可读流 */const fs = require("fs");const eventemitter = require("events");class readstream extends eventemitter { constructor(path, options = {}) { super(); //参考fs 写实例需要用到的参数 this.path = path; this.flags = options.flags || "r"; this.encoding - options.encoding || null; this.autoclose = options.autoclose || true; this.start = options.start || 0; this.end = options.end; this.highwatermark = options.highwatermark || 64 * 1024; this.fd = undefined; this.offset = this.start; this.flowing = false; this.open(); this.on("newlistener", function (type) { if (type === "data") { this.flowing = true; this.read(); } }); } destroy(err) { if (err) { this.emit("error"); } if (this.autoclose) { fs.close(this.fd, () => { this.emit("close"); }); } } open() { fs.open(this.path, this.flags, (err, fd) => { if (err) { return this.destroy(err); } this.fd = fd; this.emit("open", fd); }); } resume() { if (!this.flowing) { this.flowing = true; this.read(); } } pause() { this.flowing = false; } read() { if (typeof this.fd !== "number") { return this.once("open", () => this.read()); } let howmuchtoread = this.end ? math.min(this.end - this.offset + 1, this.highwatermark) : this.highwatermark; const buffer = buffer.alloc(this.highwatermark); fs.read( this.fd, buffer, 0, howmuchtoread, this.offset, (err, bytesread) => { if (bytesread) { this.offset += bytesread; this.emit("data", buffer.slice(0, bytesread)); if (this.flowing) { this.read(); } } else { this.emit("end"); this.destroy(); } } ); }}module.exports = readstream;
调用代码const readstream = require("./initreadstream");let rs = new readstream(apath, { flags: "r", encoding: null, //默认编码格式是buffer autoclose: true, //相当于需要调用close方法，如果为false 文件读取end的时候 就不会执行 close start: 0, highwatermark: 3, //每次读取的个数 默认是64*1024个字节});
可写流的实现待续...
更多编程相关知识，请访问：编程视频！！
以上就是浅谈nodejs中的可读流，可读流如何实现？的详细内容。