Node.js数据流Stream之Duplex流和Transform流用法

在Node.js中，数据流Stream是一种基于事件的API，用于将数据从一个地方传输到另一个地方。Stream是异步的，基于事件的，具有高效、可扩展、高吞吐量等优点。其中，Duplex流和Transform流是两种比较常用的数据流，本文将分别介绍它们的用法。

Duplex流

Duplex流是同时同时支持读取和写入的数据流。在Node.js stream模块中，Duplex流继承了Readable和Writable两个类。Duplex流可以用于一些需要同时读写的场景，如网络服务器、本地数据库等。

创建Duplex流

可以使用stream.Duplex构造函数来创建Duplex流，也可以通过继承stream.Duplex类来创建自定义的Duplex流。下面是一个简单的示例，创建一个输出a-z小写字母的Duplex流：

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

const lowerCaseAlphabetStream = new Duplex({
  read(size) {
    if (this.charCode > 122) {
      this.push(null);
      return;
    }
    const letter = String.fromCharCode(this.charCode++);
    this.push(letter);
  },
  write(chunk, encoding, callback) {
    console.log(chunk.toString().toUpperCase());
    callback();
  }
});

lowerCaseAlphabetStream.charCode = 97;

lowerCaseAlphabetStream.pipe(process.stdout);

使用Duplex流

使用Duplex流主要分为以下几个步骤：

1. 读取数据：通过监听data事件或调用read()方法来读取数据
2. 写入数据：通过调用write()方法来写入数据
3. 结束流：通过调用end()方法来结束流，发送EOF信号

下面是一个使用Duplex流来复制文件的示例：

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

const sourceFilePath = './source-file.txt';
const targetFilePath = './target-file.txt';

const duplexStream = new Duplex({
  write(chunk, encoding, callback) {
    this.push(chunk);
    callback();
  },
  read(size) {
    fs.readFile(sourceFilePath, (err, data) => {
      if (err) return this.emit('error', err);
      this.push(data);
      this.push(null);
    });
  }
});

const writeStream = fs.createWriteStream(targetFilePath);
duplexStream.pipe(writeStream);

上述示例中，我们创建了一个Duplex流，通过read()方法读取源文件数据，然后通过write()方法往下传递，最后通过pipe()方法将数据写入到目标文件。

Transform流

Transform流是Duplex流的一种特殊形式，它除了支持读写之外，还能对数据进行转换。通过Transform流的API，我们可以将数据逐一处理。Transform流常用于在不影响输入的情况下，对数据进行处理，例如压缩、解压缩、加密、字符转换等。

创建Transform流

使用stream.Transform构造函数来创建Transform流，也可以通过继承stream.Transform类创建自定义的Transform流。例如下面的示例，创建一个将输入的数据字母转为大写的Transform流：

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

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

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

使用Transform流

使用Transform流也主要有以下几个步骤：

1. 读取数据：通过监听data事件或调用read()方法来读取数据
2. 转换数据：通过transform()方法来转换数据
3. 结束流：通过调用end()方法来结束流，发送EOF信号

下面是一个使用Transform流来实现文件压缩的示例：

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

const sourceFilePath = './source-file.txt';
const targetFilePath = './target-file.txt.gz';

const compressTransformStream = new Transform({
  transform(chunk, encoding, callback) {
    zlib.gzip(chunk, (err, data) => {
      if (err) return this.emit('error', err);
      this.push(data);
      callback();
    });
  }
});

const readStream = fs.createReadStream(sourceFilePath);
const writeStream = fs.createWriteStream(targetFilePath);

readStream.pipe(compressTransformStream).pipe(writeStream);

上述示例中，我们创建了一个Transform流，在transform()方法中使用zlib模块对每个数据块进行压缩，并通过push()方法往下传递。最后我们将Transform流与文件读写流进行连接，使得文件数据经过Transform流处理后，再进行写入。

总结

Node.js中的Stream是一种强大的数据流处理工具，我们可以使用Duplex流和Transform流来处理带有读写需求的场景和需要数据转换的场景。上述示例仅仅是Stream的冰山一角，它还有很多的用法和套路，需要开发者不断实践和总结。