Node.js流

介绍

Node.js流(Stream)是一种处理大量数据的高效方式，它允许数据以小块的形式进行处理，而不必一次性加载到内存中。流在Node.js中被广泛应用于文件操作、网络通信和数据处理等场景。理解和掌握流对于编写高性能的Node.js应用至关重要，尤其是在处理大文件或高并发数据时。

原理

Node.js流的核心原理：

• 流式处理：将数据分割成小块，逐个处理，减少内存占用
• 基于事件：流是EventEmitter的实例，通过事件驱动进行数据传输
• 管道机制：通过pipe()方法将一个流的输出连接到另一个流的输入
• 四种类型：可读流(Readable)、可写流(Writable)、双工流(Duplex)和转换流(Transform)
• 背压机制：防止数据生产者速度超过消费者处理速度
• 非阻塞I/O：结合事件循环，实现高效的数据处理

图示

流的类型和关系
   ┌───────────────┐     ┌───────────────┐
   │  Readable     │────▶│  Writable     │
   │  可读流       │     │  可写流       │
   └───────────────┘     └───────────────┘
           ▲                     ▲
           │                     │
           │                     │
   ┌───────────────┐     ┌───────────────┐
   │  Duplex       │     │  Transform    │
   │  双工流       │     │  转换流       │
   └───────────────┘     └───────────────┘

流的事件和方法
- 可读流: data, end, error, close, readable; pipe(), unpipe(), resume(), pause()
- 可写流: drain, finish, error, close; write(), end()
- 双工流: 同时具有可读流和可写流的特性
- 转换流: 可读可写，且能修改或转换数据

背压机制
数据生产者 → 缓冲区 → 数据消费者
当缓冲区满时，生产者会收到信号放慢速度

实例

可读流示例

const fs = require('fs');
const path = require('path');

// 创建可读流
const readableStream = fs.createReadStream(
  path.join(__dirname, 'example.txt'),
  { encoding: 'utf8', highWaterMark: 16 * 1024 } // 16KB chunks
);

// 监听data事件
readableStream.on('data', (chunk) => {
  console.log(`接收到数据块: ${chunk.length} 字节`);
  console.log(chunk.toString());
  // 暂停流
  readableStream.pause();
  console.log('流已暂停，2秒后继续...');
  // 2秒后恢复流
  setTimeout(() => {
    readableStream.resume();
  }, 2000);
});

// 监听end事件
readableStream.on('end', () => {
  console.log('数据读取完毕');
});

// 监听error事件
readableStream.on('error', (err) => {
  console.error('读取错误:', err);
});

可写流示例

const fs = require('fs');
const path = require('path');

// 创建可写流
const writableStream = fs.createWriteStream(
  path.join(__dirname, 'output.txt'),
  { encoding: 'utf8', highWaterMark: 8 * 1024 } // 8KB chunks
);

// 写入数据
const writeData = () => {
  let canWrite = true;
  let i = 0;

  while (i < 10 && canWrite) {
    const data = `这是第 ${i + 1} 行数据\n`;
    // 返回值表示是否还能继续写入
    canWrite = writableStream.write(data);
    i++;
  }

  if (!canWrite) {
    console.log('缓冲区已满，等待drain事件...');
    // 当缓冲区清空后，继续写入
    writableStream.once('drain', writeData);
  }
};

writeData();

// 所有数据写入完成后，结束流
setTimeout(() => {
  writableStream.end('结束写入\n');
}, 1000);

// 监听finish事件
writableStream.on('finish', () => {
  console.log('数据写入完毕');
});

// 监听error事件
writableStream.on('error', (err) => {
  console.error('写入错误:', err);
});

管道和转换流示例

const fs = require('fs');
const path = require('path');
const zlib = require('zlib');
const { Transform } = require('stream');

// 创建转换流，用于处理数据
const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    // 将数据转换为大写
    const upperChunk = chunk.toString().toUpperCase();
    callback(null, upperChunk);
  }
});

// 创建读取流、压缩流和写入流
const readableStream = fs.createReadStream(
  path.join(__dirname, 'example.txt')
);

const gzip = zlib.createGzip();

const writableStream = fs.createWriteStream(
  path.join(__dirname, 'example.txt.gz')
);

// 使用管道连接流
readableStream
  .pipe(upperCaseTransform)
  .pipe(gzip)
  .pipe(writableStream)
  .on('finish', () => {
    console.log('文件已压缩并写入完成');

    // 解压文件以验证
    const gunzip = zlib.createGunzip();
    const decompressedStream = fs.createWriteStream(
      path.join(__dirname, 'example_decompressed.txt')
    );

    fs.createReadStream(path.join(__dirname, 'example.txt.gz'))
      .pipe(gunzip)
      .pipe(decompressedStream)
      .on('finish', () => {
        console.log('文件已解压完成');
      });
  });

专业解决方案

流的应用场景

• 文件操作：大文件的读写和处理
• 网络通信：HTTP请求和响应、WebSocket数据传输
• 数据处理：数据转换、压缩和解压缩
• 日志处理：实时日志收集和分析
• 媒体处理：音频和视频流的处理
• 数据库操作：大量数据的导入和导出

流的优化技巧

• 合理设置highWaterMark：根据场景调整缓冲区大小
• 使用管道(pipe)：自动处理背压，简化代码
• 避免频繁创建流：复用流对象，减少资源消耗
• 使用流式API：如mysql2的流式查询，处理大量查询结果
• 错误处理：为每个流添加错误事件监听器
• 及时销毁流：使用destroy()方法释放资源

背压处理

• 理解背压：当数据生产者速度超过消费者处理速度时产生
• 管道自动处理：pipe()方法内部实现了背压处理
• 手动处理背压：监听drain事件，控制写入速度
• 避免背压积累：及时处理数据，不要让缓冲区持续增长
• 使用Writable.writableHighWaterMark和Writable.writableLength监控缓冲区状态

自定义流实现

• 继承基础流类：Readable、Writable、Duplex、Transform
• 实现必要方法：_read()、_write()、_transform()等
• 遵循流的协议：正确触发事件，处理背压
• 使用Object Mode：处理非Buffer/字符串类型的数据
• 错误处理：在自定义流中正确处理和传播错误

流与内存管理

• 流式处理减少内存占用：适合处理大文件和大量数据
• 避免内存泄漏：及时关闭流，清理事件监听器
• 监控内存使用：使用process.memoryUsage()跟踪内存变化
• 流的暂停和恢复：根据需要控制数据流动
• 背压机制：防止内存溢出

工具推荐

• 流操作：through2、split2、concat-stream
• 文件处理：fs-extra、graceful-fs
• 压缩/解压：zlib、archiver、unzipper
• 网络流：request、got、node-fetch
• 数据库流：mysql2、mongo-stream
• 测试工具：stream-tester、tape-async
• 日志流：winston、bunyan