深入学习Node.js中的Stream：高效数据处理与管道应用

之前我们已经学习过了nodejs中的Stream的基础知识，本次课程我们学习一下如何创建自己的流。

创建自己的流

创建一个Writable Stream，新建writeable.js

const { Writable } = require("stream");
const outStream = new Writable({
  write(chunk,encoding,callback) {
    console.log(chunk.toString());
    callback();
  }
});
process.stdin.pipe(outStream);
// 保存文件为writable.js然后用node运行
// 不管你输入什么，都会得到相同的结果

创建一个Readable Stream，新建readable1.js

const {Readable} = require("stream");
const inStream = new Readable();
inStream.push("ABCDEFGHIJKLM");
inStream.push ("NOPQRSTUVWXYZ");
inStream.push(null); // No more data

inStream.on('data', (chunk)=> {
  process.stdout.write(chunk)
  console.log('写数据流')
})

//保存文件为readable.js然后用node运行
//我们先把所有数据都push进去了，然后pipe

之前的readable1.js中我们是把所有的数据push到流里面，然后监听data事件吐数据

这样不太好，我们来改成默认不吐数据，当别人要数据的时候再吐数据，只需要实现关键方法read，新建readable2.js

const {Readable} = require('stream')

const inStream = new Readable({
  read(size){
    const char = String.fromCharCode(this.currentCharCode++)
    this.push(char)
    console.log(`推了${char}`)
    if(this.currentCharCode > 90) { // Z
      this.push(null)
    }
  }
})

inStream.currentCharCode = 65 // A
inStream.pipe(process.stdout) // read会被process.stdout不停的调用

创建一个双向流

这段代码创建了一个能够从标准输入读取数据并同时将数据写入到标准输出的 Duplex 流。
在这个过程中，它通过 write 方法处理写入操作，通过 read 方法生成流的输出。

const {Duplex} = require("stream");
const inoutStream = new Duplex({
  write(chunk, encoding, callback) {
    process.stdout.write(chunk.toString());
    callback();
  },
  read(size) {
    this.push(String.fromCharCode(this.currentCharCode++));
    if (this.currentCharCode > 90) {
      this.push(null);
    }
  }
});
inoutStream.currentCharCode = 65;
process.stdin.pipe(inoutStream).pipe(process.stdout);

变换流的实现

以下代码创建了一个能够将输入的文本转换为大写字母的 Transform 流
并通过管道将输入流和输出流连接在一起，实现从标准输入到标准输出的数据流转换

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

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

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

nodejs内置的transform流，新建gzip.js

运行node gzip.js ./1.js
它会读取指定的文件，使用gzip压缩其内容，并将压缩后的数据写入一个新文件，
该文件名为原始文件名加上“.gz”扩展名

const fs = require('fs');
const zlib = require('zlib');
const file = process.argv[2];

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream(file + ".gz"))

升级进度条和完成时打打印

const fs = require('fs');
const zlib = require('zlib');
const file = process.argv[2];

fs.createReadStream(file)
  .pipe(zlib.createGzip())
// 在数据流中监听 "data" 事件，输出点号 (.) 表示进度
  .on("data", () => process.stdout.write("."))
  .pipe(fs.createWriteStream(file + ".gz"))
// 在可写流的 "finish" 事件上监听，输出 "Done" 表示完成:
  .on('finish', () => console.log("Done"))

再次升级将进度条使用变化流进行封装

const fs = require('fs');
const zlib = require('zlib');
const {Transform} = require("stream")
const file = process.argv[2];

const reportProgress = new Transform({
  transform(chunk, encoding, callback) {
    process.stdout.write(".");
    callback(null, chunk)
  }
});

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file + ".gz"))
  .on('finish', () => console.log("Done"))

加上文件加密

//...
fs.createReadStream(file)
  .pipe(crypto.createCipher("aes192", "123456"))
  .pipe(zlib.createGzip())
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file + ".gz"))
  .on('finish', () => console.log("Done"));

结尾语

Node.js中的Stream是一种高效处理数据的方式，特别适用于大规模数据的读写和传输。通过合理使用可读流、可写流、双工流和转换流，我们可以构建出复杂而高效的数据处理管道。Stream的事件机制和灵活性使得我们能够更好地控制数据的流向和处理过程。在实际应用中，合理使用Stream可以显著降低内存占用，提高程序的性能。特别是在处理大文件、网络数据传输等场景下，Stream的优势更为明显。同时，Node.js提供了丰富的内置模块和工具，例如zlib、crypto等，与Stream结合使用可以轻松实现诸如压缩、加密等功能。总体而言，深入理解和熟练使用Node.js中的Stream对于Node.js开发者是一项重要的技能。通过合理的流式数据处理，我们能够更好地应对各种场景下的数据需求，提高程序的可维护性和可扩展性。