Node.js流基础：高效处理I/O操作的核心技术

原创

qife122

发布于 2025-08-09 17:43:24

10100

代码可运行

文章被收录于专栏：网络安全技术点滴分享网络安全技术点滴分享

运行总次数：0

代码可运行

Node.js流基础

Node.js本质上是异步和事件驱动的，因此非常擅长处理I/O密集型任务。如果您的应用程序需要执行I/O操作，可以利用Node.js提供的流(Streams)功能。

关键要点

Node.js流是异步和事件驱动的，通过将数据分解为更小、更易管理的块来简化I/O操作
流可分为可读(Readable)、可写(Writable)、双工(Duplex，既可读又可写)或转换(Transform，在传输过程中修改数据)
pipe()函数是Node.js流中的有用工具，允许从源读取数据并写入目标，而无需手动管理数据流
现代Node.js提供了stream.pipeline()和stream.finished()等实用程序以及基于Promise的API，以实现更好的错误处理和流控制
流可以与async/await模式一起使用，使代码更清晰、更易维护

什么是流

Node.js中的流受到Unix管道的启发，提供了一种以流式方式从源读取数据并将其传输到目标的机制。简单地说，流就是一个EventEmitter，并实现了一些特殊方法。根据实现的方法，流可以是可读、可写、双工或转换。

可读流

可读流允许您从源读取数据。源可以是任何东西：文件系统中的简单文件、内存中的缓冲区，甚至是另一个流。

从流中读取

从流中读取数据的最佳方法是监听data事件并附加回调：

const fs = require('fs');
const readableStream = fs.createReadStream('file.txt');
let data = '';

readableStream.on('data', function(chunk) {
  data += chunk;
});

readableStream.on('end', function() {
  console.log(data);
});

readableStream.on('error', (err) => {
  console.error('Error reading stream:', err);
});

现代ECMAScript特性

我们可以使用async/await重写上面的代码：

const fs = require('fs');
const { Readable } = require('stream');
const { promisify } = require('util');

const streamToString = async (stream) => {
  const chunks = [];
  
  for await (const chunk of stream) {
    chunks.push(typeof chunk === 'string' ? chunk : chunk.toString());
  }
  
  return chunks.join('');
};

async function readFile() {
  try {
    const readableStream = fs.createReadStream('file.txt');
    const content = await streamToString(readableStream);
    console.log(content);
  } catch (err) {
    console.error('Error reading file:', err);
  }
}

readFile();

可写流

可写流允许您将数据写入目标。与可读流一样，它们也是EventEmitter，并在不同点发出各种事件。

写入流

要将数据写入可写流，您需要在流实例上调用write()：

const fs = require('fs');
const readableStream = fs.createReadStream('file1.txt');
const writableStream = fs.createWriteStream('file2.txt');

readableStream.setEncoding('utf8');

readableStream.on('data', function(chunk) {
  writableStream.write(chunk);
});

处理背压(Backpressure)

更好的方法是处理背压：

const fs = require('fs');
const readableStream = fs.createReadStream('file1.txt');
const writableStream = fs.createWriteStream('file2.txt');

readableStream.setEncoding('utf8');

readableStream.on('data', function(chunk) {
  const canContinue = writableStream.write(chunk);
  if (!canContinue) {
    readableStream.pause();
  }
});

writableStream.on('drain', function() {
  readableStream.resume();
});

readableStream.on('end', function() {
  writableStream.end();
});

readableStream.on('error', (err) => {
  console.error('Read error:', err);
  writableStream.end();
});

writableStream.on('error', (err) => {
  console.error('Write error:', err);
});

双工和转换流

双工流是可读和可写流的组合。它们维护两个独立的内部缓冲区，一个用于读取，一个用于写入，彼此独立运行。

转换流是一种特殊的双工流，可以在数据写入和读取时修改或转换数据。与双工流不同，双工流的输入和输出是分开的，而转换流的输出与其输入直接相关。

const { Transform } = require('stream');

const upperCaseTr = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

process.stdin
  .pipe(upperCaseTr)
  .pipe(process.stdout);