Node.js：流数据入门教程

❝

原文链接：Node.js – Streams^[1]，from tutorialspoint.com.

❞

❝

背景：在学习 body-parser 仓库源码^[2]时，遇到将 request 实例中的请求数据作为流数据处理的逻辑，这一块之前并没有学过，因此找来一篇国外的文章学习。值得一提的是，React 18 中也引入了 streaming server rendering 概念^[3]，进一步加深了学习流数据处理的必要性。

❞

流（Stream）是一种传递数据的方式，允许以连续的方式从源读取数据或将数据写入目标对象。Node.js 中，有 4 种类型的流：

Readable – 用于读取操作的流
Writable – 用于写操作的流
Duplex – 可同时用于读、写操作的流
Transform – Duplex 流的一种，输出（output）是基于输入（input）计算的

每种类型的流都是一个 「EventEccurrence」 实例，实例支持注册不同阶段的监听事件。常用的一些事件是包括：

data – 当还有数据可以读取时会触发这个事件
end – 当没有数据可以读取时会触发这个事件
error – 当接收或写入数据发生错误时会触发这个事件
finish – 当数据操作完成（flushed，读取或写入）时会触发这个事件

本文列举了对流常用的一些操作。下面就来介绍。

从流中读取内容

首先，创建一个名为 input.txt 的文本文件。

Tutorials Point is giving self learning content
to teach the world in simple and easy way!!!!!

再创建一个名为 readableStream.js 的文件。

const fs = require('node:fs')
let data = ''

// Create a readable stream
const readableStream = fs.createReadStream('./input.txt')

// Set the encoding to be utf8. 
readableStream.setEncoding('utf-8')

// Handle stream events -->
readableStream.on('data', (chunk) => {
  console.log('[chunk]', chunk)
  data += chunk
})

readableStream.on('end', () => {
  console.log(data)
})

readableStream.on('error', (err) => {
  console.log(err.stack)
})

console.log('Program Ended')

通过 ReadStream 对象上的 data 事件，就能一个一个 chunk 的获得读取到的数据，拼接起来后就是完整的数据了。

运行 readableStream.js，查看结果。

$ node readableStream.js

验证输出。

Program Ended
[chunk] Tutorials Point is giving self learning content
to teach the world in simple and easy way!!!!!

Tutorials Point is giving self learning content
to teach the world in simple and easy way!!!!!

写入到流

创建一个名为 writableStream.js 的文件。

const fs = require('node:fs')
let data = 'Simply Easy Learning'

// Create a writable stream
const writableStream = fs.createWriteStream('./output.txt')

// Write the data to stream with encoding to be utf8
writableStream.write(data, 'utf-8')

// Mark the end of file
writableStream.end()

// Handle stream events --> finish, and error
writableStream.on('finish', () => {
  console.log('Write completed.')
})

writableStream.on('error', (err) => {
  console.log(err.stack)
})

console.log("Program Ended");

调用 WriteStream 对象上的 .wirte() 方法，就能向 WriteStream 对象指定的目标地址上写入数据了。

运行 writableStream.js，查看结果。

$ node writableStream.js

验证输出。

Program Ended
Write completed.

执行后，会在当前目录创建一个名为 output.txt 的文件。

Simply Easy Learning

Piping the Streams

管道（Piping）是一种机制，支持将一个流的输出作为另一个流的输入。它通常用于从一个流获取数据，并将这个流的输出传递到另一个流。管道操作没有限制。下面我们写一个示例，展示从一个文件中读取并将其写入到另一个文件。

首先，创建一个名为 pipe.js 的文件，内容如下。

const fs = require('node:fs')

// Create a readable stream
const readableStream = fs.createReadStream('./input.txt')

// Create a writable stream
const writableStream = fs.createWriteStream('./output.txt')

// Pipe the read and write operations
// 读取 input.txt 内容并写入到 output.txt 文件
readableStream.pipe(writableStream)

console.log("Program Ended");

运行 pipe.js，查看结果。

$ node pipe.js

打开当前目录的 output.txt 文件，发现内容被覆盖为。

Tutorials Point is giving self learning content
to teach the world in simple and easy way!!!!!

Chaining the Streams

链接（Chaining）是一种将一个流的输出连接到另一个流并创建多个流操作链的机制。它通常用于管道操作。现在，我们将使用管道和链接的示例——首先压缩文件，然后再解压。

首先，创建一个名为 zlib01.js 的文件，内容如下。

const fs = require('node:fs')
const zlib = require('zlib')

// Compress the file input.txt to input.txt.gz
fs.createReadStream('input.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('input.txt.gz'))

console.log("File Compressed.");

运行 zlib01.js，查看结果。

$ node zlib01.js

验证输出。

File Compressed.

执行完成后，会在当前目录中创建了一个名为 input.txt.gz 的文件。

现在，再创建 zlib02.js，尝试将上述压缩文件解压（input.txt.gz –> input.txt）。

const fs = require('node:fs')
const zlib = require('zlib')

// Decompress the file input.txt.gz to input.txt
fs.createReadStream('input.txt.gz')
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream('input.txt'))

console.log("File Compressed.");