跳到主要内容

流处理

流的概念与重要性​

流(Stream)是Node.js中处理数据的一种抽象接口，它允许我们以连续的方式处理大量数据，而不必一次性将所有数据加载到内存中。在Node.js中，流被广泛应用于文件读写、网络通信、数据转换等场景。

理解和掌握流处理对于编写高性能的Node.js应用程序至关重要，尤其是在处理大型文件或高流量网络应用时。

流的类型​

Node.js中有四种基本的流类型：

  1. 可读流(Readable)：用于读取数据，如fs.createReadStream、http.IncomingMessage
  2. 可写流(Writable)：用于写入数据，如fs.createWriteStream、http.ServerResponse
  3. 双工流(Duplex)：既可读又可写的流，如net.Socket
  4. 转换流(Transform)：一种特殊的双工流，可以在写入数据的同时对数据进行转换，如zlib.createGzip、自定义的转换流

可读流​

可读流的工作模式​

可读流有两种工作模式：

  1. 流动模式(Flowing Mode)：数据自动从底层系统读取，并通过data事件发送给消费者
  2. 暂停模式(Paused Mode)：必须显式调用stream.read()方法来读取数据

可读流默认处于暂停模式。可以通过以下方式切换到流动模式：

  • 添加data事件监听器
  • 调用stream.resume()方法
  • 调用stream.pipe()方法将数据发送到可写流

可以通过以下方式切换到暂停模式：

  • 没有data事件监听器时，调用stream.pause()方法
  • 有data事件监听器时，移除所有data事件监听器
  • 调用stream.unpipe()方法移除所有管道

可读流的事件​

可读流常用的事件包括：

  • data：当流将数据块传递给消费者时触发
  • end：当流中没有更多数据可读时触发
  • error：当读取数据发生错误时触发
  • close：当流关闭时触发（不是所有流都会触发此事件）
  • readable：当流中有可读数据时触发

创建可读流的示例​

从文件创建可读流​

const fs = require('fs');

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

// 监听data事件
readableStream.on('data', (chunk) => {
  console.log(`读取到 ${chunk.length} 字节的数据`);
  console.log(chunk);
});

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

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

创建自定义可读流​

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

class CounterStream extends Readable {
  constructor(options) {
    super(options);
    this._count = 0;
    this._max = options.max || 10;
  }

  _read(size) {
    // 模拟异步数据生成
    setTimeout(() => {
      if (this._count < this._max) {
        const data = Buffer.from(`数据 ${this._count}\n`, 'utf8');
        this.push(data); // 将数据推送到流中
        this._count++;
      } else {
        this.push(null); // 表示数据已全部读取完毕
      }
    }, 100);
  }
}

// 使用自定义可读流
const counter = new CounterStream({ max: 5 });

counter.on('data', (chunk) => {
  console.log(chunk.toString());
});

counter.on('end', () => {
  console.log('计数结束');
});

可写流​

可写流的方法​

可写流常用的方法包括：

  • write(chunk[, encoding][, callback])：写入数据到流中
  • end([chunk][, encoding][, callback])：结束写入并发送结束信号
  • cork()：强制所有写入的数据都存到内部缓冲区
  • uncork()：刷新内部缓冲区中的所有数据
  • setDefaultEncoding(encoding)：设置默认的字符编码

可写流的事件​

可写流常用的事件包括：

  • drain：当流可以继续写入数据时触发（通常在write()返回false后）
  • finish：当调用end()方法后，所有数据都已写入到底层系统时触发
  • error：当写入数据发生错误时触发
  • pipe：当可读流通过pipe()方法连接到本可写流时触发
  • unpipe：当可读流通过unpipe()方法断开与本可写流的连接时触发

创建可写流的示例​

写入文件的可写流​

const fs = require('fs');

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

// 写入数据
function writeData() {
  let canContinue = true;
  let i = 0;
  
  while (i < 100 && canContinue) {
    // write()方法返回一个布尔值
    // 如果内部缓冲区已满，返回false
    // 否则返回true
    canContinue = writableStream.write(`这是第 ${i} 行数据\n`);
    i++;
  }
  
  // 如果不能继续写入，等待drain事件
  if (!canContinue) {
    writableStream.once('drain', writeData);
  } else {
    // 所有数据写入完成，结束流
    writableStream.end('写入完成\n');
  }
}

// 开始写入
writeData();

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

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

创建自定义可写流​

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

class LogStream extends Writable {
  constructor(options) {
    super(options);
  }

  _write(chunk, encoding, callback) {
    // 处理写入的数据
    console.log(`[LOG] ${chunk.toString()}`);
    // 调用callback表示写入完成
    callback();
  }
}

// 使用自定义可写流
const logger = new LogStream();

logger.write('第一条日志信息\n');
logger.write('第二条日志信息\n');
logger.end('最后一条日志信息\n');

双工流和转换流​

双工流​

双工流是同时实现了可读和可写接口的流。双工流的两端可以独立工作，读取和写入操作互不影响。

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

class EchoStream extends Duplex {
  constructor(options) {
    super(options);
  }

  _read(size) {
    // 可读端实现
  }

  _write(chunk, encoding, callback) {
    // 可写端实现：将输入的数据原样输出
    this.push(chunk);
    callback();
  }
}

// 使用双工流
const echoStream = new EchoStream();

// 监听data事件
echoStream.on('data', (chunk) => {
  console.log('接收到:', chunk.toString());
});

// 写入数据
echoStream.write('Hello World!\n');
echoStream.end();

转换流​

转换流是一种特殊的双工流，它的输出是从输入转换而来的。转换流不需要实现_read和_write方法，而是实现_transform方法。

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

class UpperCaseTransform extends Transform {
  constructor(options) {
    super(options);
  }

  _transform(chunk, encoding, callback) {
    // 将输入的数据转换为大写
    const transformedChunk = chunk.toString().toUpperCase();
    // 将转换后的数据推送到可读端
    this.push(transformedChunk);
    // 调用callback表示转换完成
    callback();
  }
}

// 使用转换流
const upperCaseTransform = new UpperCaseTransform();

// 监听data事件
upperCaseTransform.on('data', (chunk) => {
  console.log(chunk.toString());
});

// 写入数据
upperCaseTransform.write('hello world!\n');
upperCaseTransform.write('node.js streams\n');
upperCaseTransform.end();

流的管道​

管道(Pipe)是Node.js中流处理的一个核心概念，它允许我们将一个可读流的数据直接传递给一个可写流，实现数据的自动流动。

基本的管道示例​

const fs = require('fs');

// 创建可读流和可写流
const readableStream = fs.createReadStream('input.txt');
const writableStream = fs.createWriteStream('output.txt');

// 建立管道连接
readableStream.pipe(writableStream);

// 监听完成事件
writableStream.on('finish', () => {
  console.log('文件复制完成');
});

链式管道​

多个流可以通过链式管道连接在一起，形成一个处理流水线：

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

// 创建流
const readableStream = fs.createReadStream('input.txt');
const gzip = zlib.createGzip(); // 创建gzip压缩流
const writableStream = fs.createWriteStream('output.txt.gz');

// 建立链式管道
readableStream.pipe(gzip).pipe(writableStream);

// 监听完成事件
writableStream.on('finish', () => {
  console.log('文件压缩完成');
});

管道的错误处理​

管道操作中的错误需要特别注意，因为默认情况下，流的错误不会在管道中传播。我们需要为每个流单独添加错误处理：

const fs = require('fs');

const readableStream = fs.createReadStream('input.txt');
const writableStream = fs.createWriteStream('output.txt');

// 添加错误处理
readableStream.on('error', (err) => {
  console.error('读取错误:', err);
});

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

// 建立管道连接
readableStream.pipe(writableStream);

高级流处理技巧​

1. 背压处理​

背压(Backpressure)是流处理中的一个重要概念，它指的是当可写流无法及时处理可读流提供的数据时，系统如何处理这种情况。

在Node.js中，背压处理是自动进行的：当可写流的内部缓冲区接近满时，write()方法会返回false，此时可读流应该暂停数据的读取，直到可写流发出drain事件，表示可以继续写入数据。

const fs = require('fs');

const readableStream = fs.createReadStream('largefile.txt');
const writableStream = fs.createWriteStream('output.txt');

// 手动处理背压
readableStream.on('data', (chunk) => {
  const canContinue = writableStream.write(chunk);
  if (!canContinue) {
    readableStream.pause(); // 暂停读取数据
  }
});

writableStream.on('drain', () => {
  readableStream.resume(); // 恢复读取数据
});

2. 流的合并与分叉​

有时候我们需要将多个可读流的数据合并到一个可写流中，或者将一个可读流的数据分发到多个可写流中。

流的合并​

const fs = require('fs');
const { PassThrough } = require('stream');

// PassThrough是一个简单的转换流，它只是将输入的数据原样输出
const mergeStream = new PassThrough();

const stream1 = fs.createReadStream('file1.txt');
const stream2 = fs.createReadStream('file2.txt');
const output = fs.createWriteStream('merged.txt');

// 将mergeStream连接到输出流
mergeStream.pipe(output);

// 依次将输入流的数据合并到mergeStream
stream1.on('end', () => {
  stream2.pipe(mergeStream);
});

stream1.pipe(mergeStream, { end: false }); // { end: false }表示当stream1结束时，不要关闭mergeStream

流的分叉​

const fs = require('fs');
const { PassThrough } = require('stream');

const input = fs.createReadStream('input.txt');
const output1 = fs.createWriteStream('output1.txt');
const output2 = fs.createWriteStream('output2.txt');

// 创建两个PassThrough流作为分叉点
const pass1 = new PassThrough();
const pass2 = new PassThrough();

// 将输入流连接到两个PassThrough流
input.pipe(pass1);
input.pipe(pass2);

// 将PassThrough流连接到输出流
pass1.pipe(output1);
pass2.pipe(output2);

3. 对象模式的流​

默认情况下，Node.js的流是处理Buffer或字符串数据的。但是，我们也可以创建处理JavaScript对象的流，这称为对象模式(Object Mode)。

const { Readable, Writable, Transform } = require('stream');

// 创建对象模式的可读流
class ObjectReadableStream extends Readable {
  constructor(options) {
    super({ ...options, objectMode: true });
    this.data = [
      { name: '张三', age: 25 },
      { name: '李四', age: 30 },
      { name: '王五', age: 35 }
    ];
  }

  _read(size) {
    if (this.data.length > 0) {
      this.push(this.data.shift());
    } else {
      this.push(null);
    }
  }
}

// 创建对象模式的转换流
class ObjectTransformStream extends Transform {
  constructor(options) {
    super({ ...options, objectMode: true });
  }

  _transform(chunk, encoding, callback) {
    // 转换数据：添加一个属性
    chunk.isAdult = chunk.age >= 18;
    this.push(chunk);
    callback();
  }
}

// 创建对象模式的可写流
class ObjectWritableStream extends Writable {
  constructor(options) {
    super({ ...options, objectMode: true });
  }

  _write(chunk, encoding, callback) {
    console.log('处理对象:', chunk);
    callback();
  }
}

// 使用对象模式的流
const readable = new ObjectReadableStream();
const transform = new ObjectTransformStream();
const writable = new ObjectWritableStream();

readable.pipe(transform).pipe(writable);

实际应用场景​

1. 大文件处理​

流处理非常适合处理大文件，因为它不需要一次性将整个文件加载到内存中：

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

// 创建可读流
const fileStream = fs.createReadStream('large-log-file.txt', {
  encoding: 'utf8'
});

// 创建readline接口
const rl = readline.createInterface({
  input: fileStream,
  crlfDelay: Infinity
});

// 逐行处理文件
let lineCount = 0;
rl.on('line', (line) => {
  lineCount++;
  if (line.includes('ERROR')) {
    console.log(`第 ${lineCount} 行包含错误: ${line}`);
  }
});

rl.on('close', () => {
  console.log(`文件处理完成，共 ${lineCount} 行`);
});

2. HTTP请求处理​

在Web应用中，流可以用于处理HTTP请求和响应：

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

const server = http.createServer((req, res) => {
  if (req.url === '/download') {
    // 设置响应头
    res.writeHead(200, {
      'Content-Type': 'application/octet-stream',
      'Content-Disposition': 'attachment; filename="large-file.zip"'
    });
    
    // 创建可读流并通过管道发送响应
    const fileStream = fs.createReadStream('large-file.zip');
    fileStream.pipe(res);
    
    fileStream.on('error', (err) => {
      res.statusCode = 500;
      res.end('文件读取错误');
    });
  } else {
    res.statusCode = 404;
    res.end('未找到该资源');
  }
});

server.listen(3000, () => {
  console.log('服务器运行在 http://localhost:3000');
});

3. 数据转换与处理​

流可以用于数据的转换和处理，如压缩、加密、格式化等：

const fs = require('fs');
const zlib = require('zlib');
const crypto = require('crypto');

// 创建各种流
const readStream = fs.createReadStream('input.txt');
const gzip = zlib.createGzip(); // 压缩流
const cipher = crypto.createCipher('aes-256-cbc', 'secret-key'); // 加密流
const writeStream = fs.createWriteStream('output.txt.gz.enc');

// 建立链式管道
readStream
  .pipe(gzip)
  .pipe(cipher)
  .pipe(writeStream)
  .on('finish', () => {
    console.log('文件已压缩并加密');
  })
  .on('error', (err) => {
    console.error('处理错误:', err);
  });

流处理的最佳实践​

1. 始终处理错误事件​

流的错误不会自动传播，因此必须为每个流添加错误处理：

stream.on('error', (err) => {
  console.error('流错误:', err);
  // 进行错误处理，如关闭其他流、释放资源等
});

2. 使用管道代替手动数据处理​

管道可以自动处理背压，使代码更简洁、更可靠：

// 好的做法：使用管道
readableStream.pipe(writableStream);

// 不太好的做法：手动处理数据（除非有特殊需求）
readableStream.on('data', (chunk) => {
  writableStream.write(chunk);
});

3. 正确管理流的生命周期​

确保在适当的时候关闭流，以释放系统资源：

// 当不再需要流时，调用destroy方法
stream.destroy();

// 或者监听完成事件，然后进行清理
stream.on('end', () => {
  // 清理资源
});

4. 避免在流处理中阻塞事件循环​

流处理应该是非阻塞的，避免在流的回调中执行耗时的同步操作：

// 不好的做法：在data事件中执行阻塞操作
stream.on('data', (chunk) => {
  // 执行耗时的同步操作
  const result = processLargeData(chunk); // 阻塞事件循环
  writableStream.write(result);
});

// 好的做法：将耗时操作移至工作线程或拆分处理
stream.on('data', (chunk) => {
  // 启动工作线程处理数据
  processDataInWorkerThread(chunk).then((result) => {
    writableStream.write(result);
  });
});

5. 使用第三方流处理库​

对于复杂的流处理需求，可以考虑使用第三方库，如：

  • through2：简化转换流的创建
  • pump：更安全的管道处理，自动处理错误
  • concat-stream：将流的数据合并为一个缓冲区
  • mississippi：提供一系列流处理工具函数

总结​

流是Node.js中处理数据的强大机制，它允许我们以高效、内存友好的方式处理大量数据。Node.js提供了四种基本的流类型：可读流、可写流、双工流和转换流。

通过掌握流的基本概念、工作原理和高级技巧，我们可以编写出高性能、可扩展的Node.js应用程序。流处理在大文件操作、网络通信、数据转换等场景中有着广泛的应用。

在实际开发中，我们应该始终遵循流处理的最佳实践，如正确处理错误、使用管道、管理流的生命周期等，以确保应用程序的可靠性和性能。