node.js中stream流中可读流和可写流的实现与使用方法实例分析

const fs = require('fs');
//创建一个文件可读流
let rs = fs.createReadStream('./1.txt', {
  //文件系统标志
  flags: 'r',
  //数据编码，如果调置了该参数，则读取的数据会自动解析
  //如果没调置，则读取的数据会是 Buffer
  //也可以通过 rs.setEncoding() 进行设置
  encoding: 'utf8',
  //文件描述符，默认为null
  fd: null,
  //文件权限
  mode: 0o666,
  //文件读取的开始位置
  start: 0,
  //文件读取的结束位置(包括结束位置)
  end: Infinity,
  //读取缓冲区的大小，默认64K
  highWaterMark: 3
});
//文件被打开时触发
rs.on('open', function () {
  console.log('文件打开');
});
//监听data事件，会让当前流切换到流动模式
//当流中将数据传给消费者后触发
//由于我们在上面配置了 highWaterMark 为 3字节，所以下面会打印多次。
rs.on('data', function (data) {
  console.log(data);
});
//流中没有数据可供消费者时触发
rs.on('end', function () {
  console.log('数据读取完毕');
});
//读取数据出错时触发
rs.on('error', function () {
  console.log('读取错误');
});
//当文件被关闭时触发
rs.on('close', function () {
  console.log('文件关闭');
});

const fs = require('fs');
//创建一个文件可读流
let rs = fs.createReadStream('./1.txt', {
  highWaterMark: 3
});
rs.on('data', function (data) {
  console.log(`读取了 ${data.length} 字节数据 : ${data.toString()}`);
  //使流动模式的流停止触发'data'事件，切换出流动模式，数据都会保留在内部缓存中。
  rs.pause();
  //等待3秒后，再恢复触发'data'事件，将流切换回流动模式。
  setTimeout(function () {
    rs.resume();
  }, 3000);
});

const fs = require('fs');
let rs = fs.createReadStream('./1.txt', {
  highWaterMark: 1
});
//当流中有数据可供读取时就触发
rs.on('readable', function () {
  let data;
  //循环读取数据
  //参数表示要读取的字节数
  //如果可读的数据不足字节数，则返回缓冲区剩余数据
  //如是没有指定字节数，则返回缓冲区中所有数据
  while (data = rs.read()) {
    console.log(`读取到 ${data.length} 字节数据`);
    console.log(data.toString());
  }
});

const fs = require('fs');
//创建一个文件可写流
let ws = fs.createWriteStream('./1.txt', {
  highWaterMark: 3
});
//往流中写入数据
//参数一表示要写入的数据
//参数二表示编码方式
//参数三表示写入成功的回调
//缓冲区满时返回false，未满时返回true。
//由于上面我们设置的缓冲区大小为 3字节，所以到写入第3个时，就返回了false。
console.log(ws.write('1', 'utf8'));
console.log(ws.write('2', 'utf8'));
console.log(ws.write('3', 'utf8'));
console.log(ws.write('4', 'utf8'));
function writeData() {
  let cnt = 9;
  return function () {
    let flag = true;
    while (cnt && flag) {
      flag = ws.write(`${cnt}`);
      console.log('缓冲区中写入的字节数', ws.writableLength);
      cnt--;
    }
  };
}
let wd = writeData();
wd();
//当缓冲区中的数据满的时候，应停止写入数据，
//一旦缓冲区中的数据写入文件了，并清空了，则会触发 'drain' 事件，告诉生产者可以继续写数据了。
ws.on('drain', function () {
  console.log('可以继续写数据了');
  console.log('缓冲区中写入的字节数', ws.writableLength);
  wd();
});
//当流或底层资源关闭时触发
ws.on('close', function () {
  console.log('文件被关闭');
});
//当写入数据出错时触发
ws.on('error', function () {
  console.log('写入数据错误');
});

const fs = require('fs');
//创建一个文件可写流
let ws = fs.createWriteStream('./1.txt', {
  highWaterMark: 3
});
//往流中写入数据
//参数一表示要写入的数据
//参数二表示编码方式
//参数三表示写入成功的回调
//缓冲区满时返回false，未满时返回true。
//由于上面我们设置的缓冲区大小为 3字节，所以到写入第3个时，就返回了false。
console.log(ws.write('1', 'utf8'));
console.log(ws.write('2', 'utf8'));
console.log(ws.write('3', 'utf8'));
console.log(ws.write('4', 'utf8'));
//调用end()表明已经没有数据要被写入，在关闭流之前再写一块数据。
//如果传入了回调函数，则将作为 'finish' 事件的回调函数
ws.end('最后一点数据', 'utf8');
//调用 end() 且缓冲区数据都已传给底层系统时触发
ws.on('finish', function () {
  console.log('写入完成');
});

const fs = require('fs');
let ws = fs.createWriteStream('./1.txt', {
  highWaterMark: 1
});
//调用 cork() 后，会强制把所有写入的数据缓冲到内存中。
//不会因为写入的数据超过了 highWaterMark 的设置而写入到文件中。
ws.cork();
ws.write('1');
console.log(ws.writableLength);
ws.write('2');
console.log(ws.writableLength);
ws.write('3');
console.log(ws.writableLength);
//将调用 cork() 后的缓冲数据都输出到目标，也就是写入文件中。
ws.uncork();

const fs = require('fs');
let ws = fs.createWriteStream('./1.txt', {
  highWaterMark: 1
});
//调用一次 cork() 就应该写一次 uncork()，两者要一一对应。
ws.cork();
ws.write('4');
ws.write('5');
ws.cork();
ws.write('6');
process.nextTick(function () {
  //注意这里只调用了一次 uncork()
  ws.uncork();
  //只有调用同样次数的 uncork() 数据才会被输出。
  ws.uncork();
});

const fs = require('fs');
//创建一个可读流
let rs = fs.createReadStream('./1.txt', {
  highWaterMark: 3
});
//创建一个可写流
let ws = fs.createWriteStream('./2.txt', {
  highWaterMark: 3
});
rs.on('data', function (data) {
  let flag = ws.write(data);
  console.log(`往可写流中写入 ${data.length} 字节数据`);
  //如果写入缓冲区已满，则暂停可读流的读取
  if (!flag) {
    rs.pause();
    console.log('暂停可读流');
  }
});
//监控可读流数据是否读完
rs.on('end', function () {
  console.log('数据已读完');
  //如果可读流读完了，则调用 end() 表示可写流已写入完成
  ws.end();
});
//如果可写流缓冲区已清空，可以再次写入，则重新打开可读流
ws.on('drain', function () {
  rs.resume();
  console.log('重新开启可读流');
});

const fs = require('fs');
//创建一个可读流
let rs = fs.createReadStream('./1.txt', {
  highWaterMark: 3
});
//创建一个可写流
let ws = fs.createWriteStream('./2.txt', {
  highWaterMark: 3
});
let ws2 = fs.createWriteStream('./3.txt', {
  highWaterMark: 3
});
//绑定可写流到可读流，自动将可读流切换到流动模式，将可读流的所有数据推送到可写流。
rs.pipe(ws);
//可以绑定多个可写流
rs.pipe(ws2);

const fs = require('fs');
//创建一个可读流
let rs = fs.createReadStream('./1.txt', {
  highWaterMark: 3
});
//创建一个可写流
let ws = fs.createWriteStream('./2.txt', {
  highWaterMark: 3
});
let ws2 = fs.createWriteStream('./3.txt', {
  highWaterMark: 3
});
rs.pipe(ws);
rs.pipe(ws2);
//解绑可写流，如果参数没写，则解绑所有管道
setTimeout(function () {
  rs.unpipe(ws2);
}, 0);