Node.js Stream流教程：Readable与Writable流创建、模式切换及Pipe用法

Node.js Stream流概述与类型

本文通过具体实例，详细解析Node.js中Stream流机制下可读流与可写流的实现原理及使用方法，旨在为开发者提供清晰的参考指南。

在Node.js环境中，Stream（流）是处理流式数据的核心抽象接口。系统内置了多种流对象，例如HTTP模块中的request和response对象，以及process.stdout，它们均为Stream的具体实例。

根据数据流向的不同，流可分为可读（Readable）、可写（Writable）或双向（Duplex）类型。值得注意的是，所有的流对象均继承自events模块，因此它们都是EventEmitter的实例，支持事件监听机制。

一、流的类型

Node.js主要提供了四种基础流类型，分别适用于不同的数据处理场景：

1、Writable（可写流）：典型实现为fs.createWriteStream()，用于将数据写入目标资源。

2、Readable（可读流）：典型实现为fs.createReadStream()，用于从源资源读取数据。

3、Duplex（双向流）：同时具备可读和可写功能，例如net.Socket网络套接字。

4、Transform（转换流）：一种特殊的Duplex流，可在读写过程中对数据进行修改或转换，例如zlib.createDeflate()用于数据压缩。

流的数据模式与可读流控制

二、流中的数据有两种模式

1、二进制模式：在此模式下，流处理的数据单元为字符串（string）或Buffer对象。

2、对象模式：在此模式下，流内部处理的是普通的JavaScript对象，适用于需要传递复杂数据结构而非原始字节的场景。

三、可读流的两种模式

1、流动模式（flowing）：在此模式下，系统会自动从底层读取数据，并通过事件机制尽可能快地将数据推送给应用程序。

2、暂停模式（paused）：在此模式下，数据不会自动推送，必须显式调用read()方法才能从缓冲区中读取数据。

所有可读流在初始化时默认处于暂停模式。开发者可以通过以下几种方式将其切换至流动模式：

1、为流添加'data'事件监听回调。

2、直接调用resume()方法。

3、调用pipe()方法将流数据管道传输至另一个可写流。

若需将可读流从流动模式切换回暂停模式，可采用以下方法：

1、若当前没有配置管道目标（pipe target），直接调用pause()方法。

2、若已配置管道目标，需先移除所有管道目标，或调用unpipe()方法移除特定的管道目标。

可读流的事件监听与模式切换

四、创建可读流，并监听事件


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('文件关闭');
});

需要特别注意，'open'和'close'事件并非所有类型的流都会触发，具体取决于流的实现方式。

当监听'data'事件后，系统会尽可能快地读取并推送数据。但在某些场景下，可能需要暂停数据读取以执行其他操作。

此时，可以使用pause()和resume()方法来手动控制流的读取节奏。


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);
});

可读流还提供了一个'readable'事件，当缓冲区中有数据可供读取时触发该事件。

需注意，一旦监听'readable'事件，流会自动停止流动（即进入暂停模式）。此时必须调用read()方法显式读取数据，否则数据将滞留在缓冲区中。

强烈建议不要混用on('data')、on('readable')和pipe()这三种数据消费方式，否则可能导致不可预知的行为或数据丢失。


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('写入数据错误');
});

对于可写流，end()方法用于结束写入操作，并触发'finish'事件，表示所有数据已成功写入底层系统。


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('写入完成');
});

此外，cork()和uncork()方法主要用于优化性能。当需要写入大量小块数据时，内部缓冲机制可能导致性能下降，通过cork()暂停缓冲输出，待批量数据处理完毕后再调用uncork()释放缓冲，可显著提升写入效率。


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();

务必确保cork()的调用次数与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();
});

六、可读流的 pipe() 方法

pipe()方法在可读流与可写流之间建立连接，其内部逻辑类似于手动处理数据的代码实现，但更加简洁高效。


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('重新开启可读流');
});

我们可以直接使用pipe()方法完成上述复杂的数据传输功能，无需手动处理事件和缓冲。


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);

若需断开连接，可以手动调用unpipe()方法将可写流从管道中解绑。


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);

希望本文所述的Node.js Stream流实现与使用方法，能对大家的程序设计工作有所帮助。

Node.js Stream流详解：可读与可写流的实现及用法

Node.js Stream流概述与类型

流的数据模式与可读流控制

可读流的事件监听与模式切换

可写流的写入优化与管道传输

实用工具推荐