이 글에서는 Nodejs의 스트림을 소개하고 노드 판독 가능 스트림이 어떻게 구현되는지 살펴보겠습니다. 도움이 필요한 친구들이 모두 참고할 수 있기를 바랍니다.
stream의 개념
stream은 Node.js에서 스트리밍 데이터를 처리하기 위한 추상 인터페이스입니다. 스트림 모듈은 스트림 인터페이스를 구현하는 객체를 생성하는 데 사용됩니다. [추천학습: "nodejs tutorial"] stream의 역할
stream
대용량 파일을 읽고 쓰는 과정에서 한꺼번에 읽고 메모리에 쓰지는 않습니다. 매번 읽기 및 쓰기 횟수를 제어할 수 있습니다
스트림 분류
1. 읽기 가능 스트림-Readable
예: fs.createReadStream;
소스 코드 위치: lib/_stream_readable.js
2. stream -Writable
예: fs.createWriteStream;
소스 코드 위치: lib/_stream_writable.js
3. 이중 스트림-Duplex: 읽기 및 쓰기 기능을 충족합니다
예: net.Socket();
소스 코드 위치: lib/_stream_duplex.js
4. Transform stream-Transform: 목적: 압축, 트랜스코딩
예:
const { Transform } = require('stream');
Transform.call(this, '要转换的数据');//具体的使用详情 见node官网로그인 후 복사
- 소스 코드 위치: lib/_stream_tranform.js
파일을 읽을 수 있게 읽는 과정 stream
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 열기 및 닫기는 파일 스트림에 고유하게 지원됩니다. close는 파일 스트림입니다
2. 읽기 가능한 스트림에는 모두 (on('data'), on('end'), on('error'), 재개, 일시 중지가 있습니다. 이러한 메서드가 지원되는 한 읽을 수 있는 스트림
쓰기 가능한 스트림 쓰기 파일을 쓰는 과정
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로그인 후 복사
이중 스트림의 쓰기 및 읽기 과정
1. 서버(서버 코드) 구현
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. 클라이언트(클라이언트) 구현
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의 3방향 핸드쉐이크와 4방향 웨이브를 보고 싶다면 Wireshark를 사용하면 됩니다. (패킷 캡처 도구) 위의 코드를 통해 실제 프로세스를 보려면
변환 스트림의 변환 프로세스
변환 스트림은 데이터에 대해 특정 작업을 수행한 후 입력을 허용하고 출력을 반환하는 일종의 이중 스트림입니다. 두 개에는 종속성이 있습니다.
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')
rs.on('open')为断点入口进入
1、通过Stream.prototype.on.call 继承Stream类
源文件位置：no dlib/_stream_readable.js（我是通过断点点到这里 直接找，我也没找到）
- 再点进去 发现 Stream 是EventEmitter的子类 那么 可读流也可以支持发布订阅
2、监听的事件类型是否是data和readable任意一个 不是 继续 下一个事件的监听
rs.on('data')
rs.on('open')은 다음과 같습니다. 중단점 항목
2. 모니터링 이벤트 유형이 데이터인지 읽기 가능한지 여부에 관계없이 다음 이벤트에 대한 모니터링이 계속되지 않습니다
< code>rs.on('data')
data 이 부분은 두 가지 작업을 수행합니다
1. 흐름(기본값은 null)이 false가 아닌지 확인한 다음 자동으로 메서드 실행을 재개합니다. 계속 파일 읽기(여기서는 rs.pause()입니다. 흐르는 값을 수동으로 false로 설정하므로 호출을 계속하지 않습니다.)
2 그런 다음 rs.pause()를 호출하지 않으면 계속 호출합니다. 이력서에서 수행되는 작업을 확인하세요🎜🎜🎜🎜🎜🎜🎜2.1 파일 읽기를 계속하기 위해 stream.read()가 호출됩니다. ; 🎜🎜 요약: 따라서 파일 읽기가 완료될 때까지 데이터는 기본적으로 파일을 계속 읽습니다. 파일 읽기를 제어할 수 있도록 하려면 나처럼 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 并调用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个字节
});로그인 후 복사
可写流的实现
待续...
更多编程相关知识，请访问：编程视频！！
위 내용은 Nodejs의 읽기 가능한 스트림에 대한 간략한 설명입니다. 읽기 가능한 스트림을 구현하는 방법은 무엇입니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
