NodeJS 中Stream 的基本使用

NodeJS中Stream是一种非常重要的数据处理工具，它可以帮助我们高效地处理大量数据，在文件读写、网络传输等多个场景下都有广泛应用。下面我们来详细讲解NodeJS中Stream的基本使用。

什么是Stream

流（Stream）是Node.js中处理流式数据的一个抽象接口。Stream有四种类型：Readable、Writable、Duplex、Transform。在Stream中，数据是以块（chunk）为单位流动的，而不是一次性将所有数据读取到内存中。这使得在读写大量数据时不会因为内存不足而出现问题。

在Node.js中，大多数的核心API都提供了Stream的支持，比如文件读写和网络传输等。除此之外，我们也可以自定义Stream来满足特定的需求。

Stream的基本用法

创建Readable Stream

使用fs.createReadStream()来创建一个Readable Stream来读取文件的内容。

const fs = require('fs');

const readStream = fs.createReadStream('./example.txt');

readStream.on('data', (chunk) => {
  console.log(chunk.toString());  // 输出读取到的数据块
});

readStream.on('end', () => {
  console.log('读取完成');
});

readStream.on('error', (err) => {
  console.error('发生错误', err);
});

上面的代码中，我们通过fs.createReadStream()方法读取了当前目录下的example.txt文件，并将其转换为一个可读流（Readable Stream）。然后，我们通过readStream.on('data', ...)监听了data事件，当数据块可读取时触发该事件，我们将读取到的数据块输出到控制台。readStream.on('end', ...)监听end事件，当所有数据块读取完毕时触发该事件，我们输出"读取完成"的信息。readStream.on('error', ...)监听error事件，当读取过程中发生了错误时触发该事件，我们将错误信息输出到控制台。

我们也可以通过继承Readable类来创建自定义的可读流。

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

class MyReadable extends Readable {
  constructor(options) {
    super(options);
    this.data = ['a', 'b', 'c'];
  }

  _read() {
    if (!this.data.length) {
      this.push(null);
    } else {
      this.push(this.data.shift());
    }
  }
}

const myReadable = new MyReadable();

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

myReadable.on('end', () => {
  console.log('读取完成');
});

myReadable.on('error', (err) => {
  console.error('发生错误', err);
});

上面的代码中，我们通过继承Readable类创建了一个自定义的可读流MyReadable，并在构造函数中初始化了一个字符串数组data来表示数据源。我们重写了MyReadable的_read()方法，每次调用该方法时，我们将data数组中的一个元素推到流中，推完后从数组中删除该元素。当没有数据可以推送时，我们通过调用this.push(null)来告诉流读取结束。

创建完成后，我们通过myReadable.on('data', ...)监听data事件，当数据块可读取时触发该事件，我们将读取到的数据块输出到控制台。myReadable.on('end', ...)监听end事件，当所有数据块读取完毕时触发该事件，我们输出"读取完成"的信息。myReadable.on('error', ...)监听error事件，当读取过程中发生了错误时触发该事件，我们将错误信息输出到控制台。

创建Writable Stream

使用fs.createWriteStream()来创建一个Writable Stream来写入文件的内容。

const fs = require('fs');

const writeStream = fs.createWriteStream('./example.txt');

writeStream.write('Hello, World!\n');
writeStream.write('Hello, Node.js!\n');
writeStream.end('End of file.');

writeStream.on('finish', () => {
  console.log('写入完成');
});

writeStream.on('error', (err) => {
  console.error('发生错误', err);
});

上面的代码中，我们通过fs.createWriteStream()方法创建可写流（Writable Stream）并指定要写入的文件。然后，我们通过writeStream.write()方法向流中写入一些数据，并在最后通过writeStream.end()方法告诉流写入结束。注意，如果不调用writeStream.end()方法，流将不会结束，也就不会触发finish事件。最后，我们通过writeStream.on('finish', ...)监听finish事件，当所有数据块写入完成时触发该事件，我们输出"写入完成"的信息。writeStream.on('error', ...)监听error事件，当写入过程中发生了错误时触发该事件，我们将错误信息输出到控制台。

我们也可以通过继承Writable类来创建自定义的可写流。

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

class MyWritable extends Writable {
  constructor(options) {
    super(options);
    this.data = '';
  }

  _write(chunk, encoding, callback) {
    this.data += chunk.toString();
    callback();
  }
}

const myWritable = new MyWritable();

myWritable.write('Hello, World!\n');
myWritable.write('Hello, Node.js!\n');
myWritable.end('End of file.');

myWritable.on('finish', () => {
  console.log(myWritable.data);
  console.log('写入完成');
});

myWritable.on('error', (err) => {
  console.error('发生错误', err);
});

上面的代码中，我们通过继承Writable类创建了一个自定义的可写流MyWritable。在构造函数中，我们初始化了一个字符串data来表示写入的内容。我们重写了MyWritable的_write()方法，该方法会在每次写入数据时被调用，我们将写入的数据块转换为字符串，并拼接到data字符串中。当所有数据块写入完毕时，我们通过调用callback()告诉流本次写入完成。最后，我们通过myWritable.on('finish', ...)监听finish事件，当所有数据块写入完成时触发该事件，我们输出写入的内容和"写入完成"的信息。myWritable.on('error', ...)监听error事件，当写入过程中发生了错误时触发该事件，我们将错误信息输出到控制台。

结语

以上就是NodeJS中Stream的基本使用方法。通过学习这些内容，我们可以使用Stream高效地处理大量数据。如果你还想进一步了解Stream的详细内容和使用方法，可以参考Node.js官方文档。