Node.js中你不可不精的Stream（流）

var Stream = require('stream') //stream 模块引入方式

var Readable = Stream.Readable //可读的流
var Writable = Stream.Writable //可写的流
var Duplex = Stream.Duplex //可读写的流
var Transform = Stream.Transform //在读写过程中可以修改和变换数据的 Duplex 流

const fs = require('fs')

fs.readFile(file, function (err, content) { //读出来的content是Buffer
 console.log(content)
 console.log(content.toString())
})

<Buffer 64 74 09 75 61 09 63 6f 75 6e 74 0a 0a 64 74 09 75 61 09 63 6f 75 6e 74 0a 32 30 31 35 31 32 30 38 09 4d 6f 7a 69 6c 6c 61 2f 35 2e 30 20 28 63 6f 6d ... >
buffer.js:382
 throw new Error('toString failed');
 ^

Error: toString failed
 at Buffer.toString (buffer.js:382:11)

var fs = require('fs')
fs.createReadStream(bigFile).pipe(process.stdout) 

const http = require('http');
const fs = require('fs');
 
http.createServer((req, res) => {
 fs.readFile(videoPath, (err, data) => {
 res.end(data);
});
}).listen(8080);

const http = require('http');
const fs = require('fs');
 
http.createServer((req, res) => {
 fs.createReadStream(videoPath).pipe(res);
}).listen(8080);

//文件名 ReadStream.js
let fs = require('fs');//读取文件
let EventEmitter = require('events');
class ReadStream extends EventEmitter {//流操作都是基于事件的
 constructor(path, options = {}) {
 super();
 //需要的参数
 this.path = path;//读取文件的路径
 this.highWaterMark = options.highWaterMark || 64 * 1024;//缓冲区大小，默认64KB
 this.autoClose = options.autoClose || true;//是否需要自动关闭文件描述符，默认为true
 this.start = options.start || 0; //options 可以包括 start 和 end 值，使其可以从文件读取一定范围的字节而不是整个文件
 this.pos = this.start; // 从文件的那个位置开始读取内容，pos会随着读取的位置而改变
 this.end = options.end || null; // null表示没传递
 this.encoding = options.encoding || null;
 this.flags = options.flags || 'r';//以何种方式操作文件

 // 参数的问题
 this.flowing = null; // 默认为非流动模式
 // 建一个buffer存放读出来的数据
 this.buffer = Buffer.alloc(this.highWaterMark);
 this.open(); 
 // {newListener:[fn]}
 // 次方法默认同步调用的
 this.on('newListener', (type) => { // 等待着 它监听data事件
  if (type === 'data') {//当监听到data事件时，把流设置为流动模式
  this.flowing = true;
  this.read();// 开始读取 客户已经监听了data事件
  }
 })
 }
 pause(){//将流从flowing模式切换为paused模式
 this.flowing = false;
 }
 resume(){//将流从paused模式切换为flowing模式
 this.flowing =true;
 this.read();//将流从paused模式切换为flowing模式后，继续读取文件内容
 }
 read(){ // 默认第一次调用read方法时还没有获取fd，文件的打开是异步的，所以不能直接读
 if(typeof this.fd !== 'number'){ //如果fd不是number类型，证明文件还没有打开，此时需要监听一次open事件，因为文件一打开，就会触发open事件，这个在this.open()里写了
  return this.once('open',() => this.read()); // 等待着触发open事件后fd肯定拿到了，拿到以后再去执行read方法
 }
 // 当获取到fd时 开始读取文件了
 // 第一次应该读2个 第二次应该读2个
 // 第二次pos的值是4 end是4
 // 读取文件里一共4有个数为123 4，我们读取里面的123 4
 let howMuchToRead = this.end"htmlcode">

// 流：有序的有方向的，可以自己控制速率
// 读：读是将内容读取到内存中 
// 写：写是将内存或者文件的内容写入到文件内
// 读取的时候默认读 默认一次读取64k，encoding 读取出来的内容默认都是buffer
//let fs = require('fs');
//let rs = fs.createReadStream({...});//原生实现可读流
let ReadStream = require('./ReadStream');
let rs = new ReadStream('./2.txt', {
 highWaterMark: 3, // 字节
 flags:'r',//读文件
 autoClose:true, // 默认读取完毕后自动关闭文件描述符
 start:0,
 //end:3,// 流是闭合区间 包start也包end
 encoding:'utf8'
});
// 默认创建一个流 是非流动模式（上述源码中有写的），默认不会读取数据
// 如果我们需要接收数据，那我们要监听data事件，这样数据会自动的流出来
rs.on('error',function (err) {// 通常，这会在底层系统内部出错从而不能产生数据，或当流的实现试图传递错误数据时发生。
 console.log(err)
});
rs.on('open',function () {//文件被打开了，获取到了fd。内部会自动的触发这个事件 rs.emit('data'); 
 console.log('文件打开了');
});
rs.on('data',function (data) {//有数据流出来了
 console.log(data);
 rs.pause(); // 暂停触发on('data')事件，将流动模式又转化成了非流动模式
});
setTimeout(()=>{rs.resume()},3000);//三秒钟之后再将非流动模式转化为流动模式
rs.on('end',function () {// 读取完毕
 console.log('读取完毕了');
});
rs.on('close',function () {//close 事件将在流或其底层资源（比如一个文件）关闭后触发。close 事件触发后，该流将不会再触发任何事件。
 //console.log('关闭')
});


四、可写流（Writable Stream）


可写流是对数据流向设备的抽象，用来消费上游流过来的数据，通过可写流程序可以把数据写入设备，常见的是本地磁盘文件或者 TCP、HTTP 等网络响应。


常见的可写流：

  
HTTP requests, on the client
  
HTTP responses, on the server
  
fs write streams
  
zlib streams
  
crypto streams
  
TCP sockets
  
child process stdin
  
process.stdout, process.stderr

所有 Writable 流都实现了 stream.Writable 类定义的接口。


可写流的使用


调用可写流实例的 write() 方法就可以把数据写入可写流




const fs = require('fs');
const rs = fs.createReadStream(sourcePath);
const ws = fs.createWriteStream(destPath);
 
rs.setEncoding('utf-8'); // 设置编码格式
rs.on('data', chunk => {
ws.write(chunk); // 写入数据
});


监听了可读流的data事件就会使可读流进入流动模式，我们在回调事件里调用了可写流的 write() 方法，这样数据就被写入了可写流抽象的设备destPath中。


write() 方法有三个参数

  
chunk {String| Buffer}，表示要写入的数据
  
encoding 当写入的数据是字符串的时候可以设置编码
  
callback 数据被写入之后的回调函数

drain事件


如果调用 stream.write(chunk)方法返回false，表示当前缓存区已满，流将在适当的时机（缓存区清空后）触发drain事件。




const fs = require('fs');
const rs = fs.createReadStream(sourcePath);
const ws = fs.createWriteStream(destPath);
 
rs.setEncoding('utf-8'); // 设置编码格式
rs.on('data', chunk => {
let flag = ws.write(chunk); // 写入数据
if (!flag) { // 如果缓存区已满暂停读取
rs.pause();
}
});
 
ws.on('drain', () => {
rs.resume(); // 缓存区已清空 继续读取写入
});


fs.createWriteStream(path[, options])源码实现




// 文件 WriteStream.js
let fs = require('fs');
let EventEmitter = require('events');
class WriteStream extends EventEmitter {
 constructor(path, options = {}) {
 super();
 this.path = path;
 this.flags = options.flags || 'w';
 this.encoding = options.encoding || 'utf8';
 this.start = options.start || 0;
 this.pos = this.start;
 this.mode = options.mode || 0o666;
 this.autoClose = options.autoClose || true;
 this.highWaterMark = options.highWaterMark || 16 * 1024;
 this.open(); // fd 异步的 //触发一个open事件,当触发open事件后fd肯定就存在了

 // 写文件的时候 需要的参数有哪些
 // 第一次写入是真的往文件里写
 this.writing = false; // 默认第一次就不是正在写入
 // 用简单的数组来模拟一下缓存
 this.cache = [];
 // 维护一个变量，表示缓存的长度
 this.len = 0;
 // 是否触发drain事件
 this.needDrain = false;
 }
 clearBuffer() {
 let buffer = this.cache.shift();
 if (buffer) { // 如果缓存里有
  this._write(buffer.chunk, buffer.encoding, () => this.clearBuffer());
 } else {// 如果缓存里没有了
  if (this.needDrain) { // 需要触发drain事件
  this.writing = false; // 告诉下次直接写就可以了 不需要写到内存中了
  this.needDrain = false;
  this.emit('drain');
  }
 }
 }
 _write(chunk, encoding, clearBuffer) { // 因为write方法是同步调用的此时fd还没有获取到，所以等待获取到再执行write操作
 if (typeof this.fd != 'number') {
  return this.once('open', () => this._write(chunk, encoding, clearBuffer));
 }
 fs.write(this.fd, chunk, 0, chunk.length, this.pos, (err, byteWritten) => {
  this.pos += byteWritten;
  this.len -= byteWritten; // 每次写入后就要在内存中减少一下
  clearBuffer(); // 第一次就写完了
 })
 }
 write(chunk, encoding = this.encoding) { // 客户调用的是write方法去写入内容
 // 要判断 chunk必须是buffer或者字符串 为了统一，如果传递的是字符串也要转成buffer
 chunk = Buffer.isBuffer(chunk) "htmlcode">

// 可写流有缓存区的概念
// 1.第一次写入是真的向文件里写，第二次在写入的时候是放到了缓存区里
// 2.写入时会返回一个boolean类型，返回为false时表示缓存区满了，不要再写入了
// 3.当内存和正在写入的内容消耗完后，会触发一个drain事件
//let fs = require('fs');
//let rs = fs.createWriteStream({...});//原生实现可写流
let WS = require('./WriteStream')
let ws = new WS('./2.txt', {
 flags: 'w', // 写入文件，默认文件不存在会创建
 highWaterMark: 1, // 设置当前缓存区的大小
 encoding: 'utf8', // 文件里存放的都是二进制
 start: 0,
 autoClose: true, // 自动关闭文件描述符
 mode: 0o666, // 可读可写
});
// drain的触发时机，只有当highWaterMark填满时，才可能触发drain
// 当嘴里的和地下的都吃完了，就会触发drain方法
let i = 9;
function write() {
 let flag = true;
 while (flag && i >= 0) {
 i--;
 flag = ws.write('111'); // 987 // 654 // 321 // 0
 console.log(flag)
 }
}
write();
ws.on('drain', function () {
 console.log('dry');
 write();
});


总结


stream(流)分为可读流（flowing mode和paused mode）、可写流、可读写流，Node.js 提供了多种流对象。 例如， HTTP 请求 和 process.stdout 就都是流的实例。stream 模块提供了基础的 API 。使用这些 API 可以很容易地来构建实现流接口的对象。它们底层都调用了stream模块并进行封装。
好了，以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对的支持。