


A brief discussion on readable streams in Nodejs. How to implement readable streams?
This article will introduce you to the stream in Nodejs and see how Node readable streams are implemented. It has certain reference value. Friends in need can refer to it. I hope it will be helpful to everyone.
The concept of stream
Stream is an abstract interface for processing streaming data in Node.js. The stream module is used to construct objects that implement the stream interface. [Recommended learning: "nodejs Tutorial"]
The role of stream
In the process of reading and writing large files, it will not be read all at once written to memory. You can control the number of reads and writes each time
Classification of streams
1. Readable stream-Readable
Example: fs.createReadStream;
Source code location: lib/_stream_readable.js
2. Writable stream-Writable
Example: fs.createWriteStream;
Source code location: lib/_stream_writable .js
3. Duplex stream-Duplex: meets the functions of reading and writing
Example: net.Socket();
Source code location: lib/_stream_duplex.js
4. Transform stream-Transform: Purpose: compression, transcoding
Example:
const { Transform } = require('stream'); Transform.call(this, '要转换的数据');//具体的使用详情 见node官网
-Source code location: lib/_stream_tranform.js
Readable stream reading file process
- Reading file code process
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);
- Digression: I want to talk about file streams and ordinary readable streams The difference between read streams
1. Open and close are unique to file streams. Supporting open and close is a file stream
2. Readable streams are available (on('data '), on('end'), on('error'), resume, pause; so as long as these methods are supported, it is a readable stream
The process of writing a file by a writable stream
- Writing file code process
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 });//true flag = ws.write("1"); console.log({ flag });//true flag = ws.write("1"); console.log({ flag });//false
Duplex stream writing and reading process
- Write An example of a local service
1. Server (server code) implementation
const net = require("net"); //net 模块是 node自己封装的tcp层 //socket 就是双工流 能读能写 http源码就是用net模块写的 基于tcp const 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 (client) implementation
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. If you want to digress To see the three-way handshake and four-way wave of TCP, you can use wireshark (a packet capture tool) to see the actual process through my above code
Transformation flow transform process
The transformation flow is A type of duplex stream that allows input and returns output after performing certain operations on the data. The two have dependencies.
- Code process (My reference for this example )
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);
- Print results
Readable stream implementation
Follow the breakpoints and first understand the calling process of the readable stream
The previous code for the reading process of the readable stream file is used as an example breakpoint
rs.on('open')
rs.on('open') is the breakpoint entry to enter
1. Inherit the Stream class through Stream.prototype.on.call
Source file location: no dlib/_stream_readable.js (I looked here directly through the breakpoint, but I couldn’t find it either)
- Then click in and find that Stream is a subclass of EventEmitter. Then the readable stream can also support publish and subscribe
2. Whether the monitored event type is either data or readable and does not continue to monitor the next event
rs.on('data')
-
The data part does two things
1. Determine whether flowing (default value is null) is not false The automatic resume method is executed to continue file reading (my case here is rs.pause(); manually set the flowing value to false so it will not continue to be called)
2. Then if I do not call rs.pause () will continue to call resume to see what is done in resume
2.1 Finally, stream.read() is called to continue reading the file; until the file After reading, go to the emit end and close events in sequence
Summary: So data will continue to read the file by default until the file is read. If you want the file reading to be controllable, you can do the same as me. Use rs.pause()
to implement it yourself
Implementation ideas
继承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 并调用read
resume() { 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个字节 });
可写流的实现
待续...
更多编程相关知识,请访问:编程视频!!
The above is the detailed content of A brief discussion on readable streams in Nodejs. How to implement readable streams?. For more information, please follow other related articles on the PHP Chinese website!

The main uses of JavaScript in web development include client interaction, form verification and asynchronous communication. 1) Dynamic content update and user interaction through DOM operations; 2) Client verification is carried out before the user submits data to improve the user experience; 3) Refreshless communication with the server is achieved through AJAX technology.

Understanding how JavaScript engine works internally is important to developers because it helps write more efficient code and understand performance bottlenecks and optimization strategies. 1) The engine's workflow includes three stages: parsing, compiling and execution; 2) During the execution process, the engine will perform dynamic optimization, such as inline cache and hidden classes; 3) Best practices include avoiding global variables, optimizing loops, using const and lets, and avoiding excessive use of closures.

Python is more suitable for beginners, with a smooth learning curve and concise syntax; JavaScript is suitable for front-end development, with a steep learning curve and flexible syntax. 1. Python syntax is intuitive and suitable for data science and back-end development. 2. JavaScript is flexible and widely used in front-end and server-side programming.

Python and JavaScript have their own advantages and disadvantages in terms of community, libraries and resources. 1) The Python community is friendly and suitable for beginners, but the front-end development resources are not as rich as JavaScript. 2) Python is powerful in data science and machine learning libraries, while JavaScript is better in front-end development libraries and frameworks. 3) Both have rich learning resources, but Python is suitable for starting with official documents, while JavaScript is better with MDNWebDocs. The choice should be based on project needs and personal interests.

The shift from C/C to JavaScript requires adapting to dynamic typing, garbage collection and asynchronous programming. 1) C/C is a statically typed language that requires manual memory management, while JavaScript is dynamically typed and garbage collection is automatically processed. 2) C/C needs to be compiled into machine code, while JavaScript is an interpreted language. 3) JavaScript introduces concepts such as closures, prototype chains and Promise, which enhances flexibility and asynchronous programming capabilities.

Different JavaScript engines have different effects when parsing and executing JavaScript code, because the implementation principles and optimization strategies of each engine differ. 1. Lexical analysis: convert source code into lexical unit. 2. Grammar analysis: Generate an abstract syntax tree. 3. Optimization and compilation: Generate machine code through the JIT compiler. 4. Execute: Run the machine code. V8 engine optimizes through instant compilation and hidden class, SpiderMonkey uses a type inference system, resulting in different performance performance on the same code.

JavaScript's applications in the real world include server-side programming, mobile application development and Internet of Things control: 1. Server-side programming is realized through Node.js, suitable for high concurrent request processing. 2. Mobile application development is carried out through ReactNative and supports cross-platform deployment. 3. Used for IoT device control through Johnny-Five library, suitable for hardware interaction.

I built a functional multi-tenant SaaS application (an EdTech app) with your everyday tech tool and you can do the same. First, what’s a multi-tenant SaaS application? Multi-tenant SaaS applications let you serve multiple customers from a sing


Hot AI Tools

Undresser.AI Undress
AI-powered app for creating realistic nude photos

AI Clothes Remover
Online AI tool for removing clothes from photos.

Undress AI Tool
Undress images for free

Clothoff.io
AI clothes remover

AI Hentai Generator
Generate AI Hentai for free.

Hot Article

Hot Tools

MinGW - Minimalist GNU for Windows
This project is in the process of being migrated to osdn.net/projects/mingw, you can continue to follow us there. MinGW: A native Windows port of the GNU Compiler Collection (GCC), freely distributable import libraries and header files for building native Windows applications; includes extensions to the MSVC runtime to support C99 functionality. All MinGW software can run on 64-bit Windows platforms.

Notepad++7.3.1
Easy-to-use and free code editor

WebStorm Mac version
Useful JavaScript development tools

Dreamweaver Mac version
Visual web development tools

SublimeText3 Mac version
God-level code editing software (SublimeText3)