Node.js Stream

Stream（流）在node.js中是一个处理流数据的抽象接口，它允许你以流的方式读写数据，非常适合处理大量数据而不需要占用大量内存的场景。以下是Node.js中Stream的详细教程。

一、Stream的基本概念

Stream是一种抽象接口，用于表示数据的流动。在Node.js中，Stream对象实现了这个接口，允许你以流的方式读写数据。Stream的主要优势在于它可以逐块处理数据，从而高效地处理文件、网络通信等场景。

二、Stream的类型

Node.js中主要有四种类型的Stream：

1. Readable（可读流）：用于从数据源读取数据。例如，文件读取流和HTTP响应流。
2. Writable（可写流）：用于将数据写入目的地。例如，文件写入流和HTTP请求流。
3. Duplex（双工流）：既是可读流又是可写流。例如，TCP套接字。
4. Transform（变换流）：在读写过程中可以修改和变换数据。例如，zlib压缩/解压缩流。

三、Stream的基本使用

1. 创建可读流

通常使用fs.createReadStream()方法从文件中创建可读流。

const fs = require('fs');
const readableStream = fs.createReadStream('input.txt');

readableStream.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
  console.log(chunk.toString());
});

readableStream.on('end', () => {
  console.log('There is no more data to read.');
});

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

2. 创建可写流

通常使用fs.createWriteStream()方法创建文件写入流。

const fs = require('fs');
const writableStream = fs.createWriteStream('output.txt');
const data = 'Hello, this is a test!';

writableStream.write(data, 'utf8', (err) => {
  if (err) {
    console.error('Error writing to file:', err);
  } else {
    console.log('Data written to file successfully.');
  }
});

writableStream.end((err) => {
  if (err) {
    console.error('Error ending the write stream:', err);
  } else {
    console.log('Write stream has ended.');
  }
});

writableStream.on('finish', () => {
  console.log('All data has been written to the file.');
});

writableStream.on('error', (err) => {
  console.error('Error writing to file:', err);
});

3. 管道（Pipe）

管道提供了一种将可读流的数据直接传输到可写流的机制。通过调用可读流的pipe()方法，并将可写流作为参数传入，即可实现数据的管道化传输。

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

readableStream.pipe(writableStream);
console.log('Data is being piped from input.txt to output.txt.');

四、Stream的高级用法

1. 暂停与恢复

可读流可以在流动模式和暂停模式之间切换。在流动模式下，数据会自动从流中读取并触发data事件；在暂停模式下，需要显式调用stream.read()方法来读取数据。

const fs = require('fs');
const readableStream = fs.createReadStream('input.txt', { highWaterMark: 16 });

readableStream.on('readable', () => {
  let chunk;
  while (null !== (chunk = readableStream.read())) {
    console.log(`Received ${chunk.length} bytes of data.`);
    console.log(chunk.toString());
  }
});

readableStream.on('end', () => {
  console.log('There is no more data to read.');
});

// 暂停流
readableStream.pause();

// 恢复流
// readableStream.resume();

2. 变换流（Transform）

变换流可以在读写过程中对数据进行修改和变换。例如，可以使用zlib模块创建压缩/解压缩流。

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

// 压缩文件
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.txt.gz'));

console.log('File compression completed.');

// 解压文件
fs.createReadStream('input.txt.gz')
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream('input_unzipped.txt'));

console.log('File decompression completed.');

3. 链式流

通过将多个流连接起来，可以创建复杂的数据处理管道。每个流都可以对数据进行特定的处理，然后将数据传递给下一个流。

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

// 压缩并写入文件
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input_compressed.txt.gz'));

console.log('Chained compression and write completed.');

五、Stream的应用场景

1. 文件处理：使用Stream可以高效地读取和写入大文件，而不需要占用大量内存。
2. 网络通信：在处理HTTP请求和响应时，Stream允许你以流的方式发送和接收数据。
3. 实时数据处理：在处理实时数据流（如视频流、音频流）时，Stream可以确保数据的实时性和连续性。

六、Stream的错误处理与资源管理

1. 错误处理：始终监听Stream的error事件，以便在发生错误时能够及时处理。

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

2. 资源管理：使用Stream时，要确保在适当的时候关闭流以释放资源。对于文件流来说，可以在end或finish事件触发后关闭流；对于网络流来说，可以在不再需要时调用destroy()方法销毁流。

stream.on('finish', () => {
  stream.close();
});

七、性能优化

根据具体应用场景调整Stream的高水位线（highWaterMark）和其他相关参数，以优化性能。

const readableStream = fs.createReadStream('input.txt', { highWaterMark: 64 * 1024 }); // 64KB缓冲区

通过以上教程，你应该能够掌握Node.js中Stream的基本概念、类型、基本使用、高级用法以及应用场景和最佳实践。Stream是Node.js中处理数据流动的强大工具，掌握它对于开发高效、可扩展的Node.js应用程序至关重要。

本文地址：https://www.tides.cn/p_node-stream