Node.js中的Buffer和Stream

Node.js中的Buffer和Stream

计算机只能理解二进制数据，即0和1形式的数据。这些数据的顺序移动称为流。以称为块（chunk）的破碎部分流式传输数据；计算机一收到数据块就开始处理数据，而不用等待整个数据。

我们这篇文章就将讲解一下Stream和Buffer。有时，处理速度小于接收块的速率或快于接收块的速率；在这两种情况下，都需要保存块，因为处理需要最少量的块，这是使用chunk完成的。

Buffer

Buffer是一种抽象，允许我们处理 Node.js 中的原始二进制数据。它们在处理文件和网络或一般 I/O 时特别有用。

缓冲区代表分配给我们计算机的一块内存。缓冲区的大小一旦设置就无法更改。缓冲区用于存储字节。

让我们用一些数据创建一些缓冲区：

// buffer-data.js

// 创建一些缓冲区
const bufferFromString = Buffer.from('Ciao human')
const bufferFromByteArray = Buffer.from([67, 105, 97, 111, 32, 104, 117, 109, 97, 110])
const bufferFromHex = Buffer.from('4369616f2068756d616e', 'hex')
const bufferFromBase64 = Buffer.from('Q2lhbyBodW1hbg==', 'base64')

// 数据以二进制格式存储
console.log(bufferFromString) // 
console.log(bufferFromByteArray) // 
console.log(bufferFromHex) // 
console.log(bufferFromBase64) // 

// 原始缓冲区数据可以“可视化”为字符串、十六进制或 base64 
console.log(bufferFromString.toString('utf-8')) // Ciao human (默认'utf-8')
console.log(bufferFromString.toString('hex')) // 4369616f2068756d616e
console.log(bufferFromString.toString('base64')) // Q2lhbyBodW1hbg==

// 获取buffer的长度
console.log(bufferFromString.length) // 10
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

现在，让我们创建一个 Node.js 脚本，使用缓冲区将文件从一个位置复制到另一个位置：

// buffer-copy.js

import {
  readFile,
  writeFile
} from 'fs/promises'

async function copyFile (src, dest) {
  // 读取整个文件内容
  const content = await readFile(src)
  // 将该内容写入其他地方
  return writeFile(dest, content)
}

// `src` 是来自 cli 的第一个参数，`dest` 是第二个
const [src, dest] = process.argv

// 开始复制并处理结果
copyFile(src, dest)
  .then(() => console.log(`${src} copied into ${dest}`))
  .catch((err) => {
    console.error(err)
    process.exit(1)
  })
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

可以按如下方式使用此脚本：

node ./buffer-copy.js <source-file> <dest-file>
1

但是我们有没有想过当尝试复制大文件（比如说 3Gb）时会发生什么？

发生的情况是，我们会看到脚本严重失败并出现以下错误：

RangeError [ERR_FS_FILE_TOO_LARGE]: File size (3221225472) is greater than 2 GB
    at readFileHandle (internal/fs/promises.js:273:11)
    at async copyFile (file:///...//buffer-copy.js:8:19) {
  code: 'ERR_FS_FILE_TOO_LARGE'
}
1
2
3
4
5

为什么会发生这种情况？

本质上是因为当我们使用fs.readFile时，我们使用Buffer对象从内存中的文件加载所有二进制内容。根据设计，缓冲区在内存中的大小受到限制。

可以使用以下代码创建具有最大允许大小的缓冲区：

// biggest-buffer.js

import buffer from 'buffer'

// 这将分配几 GB 内存
const biggestBuffer = Buffer.alloc(buffer.constants.MAX_LENGTH) // 创建一个具有最大可能大小的缓冲区
console.log(biggestBuffer) // 
1
2
3
4
5
6
7

在某种程度上，我们可以将流视为一种抽象，它允许我们处理在不同时刻到达的数据部分（块）。每个块都是一个Buffer实例。

Stream

Stream是 Node.js 中处理流数据的抽象接口。Node.js中stream模块提供了用于实现流接口的 API。Node.js 提供了许多流对象。例如，对 HTTP 服务器的请求和process.stdout都是流实例。

我们需要 Node.js 中的流来处理和操作流数据，例如视频、大文件等。Node.js 中的 stream 模块用于管理所有流。流是一个抽象接口，用于与 Node.js 中的流数据一起工作。Node.js 为我们提供了许多流对象。

例如，如果我们请求HTTP 服务器和进程，则两者都被视为流实例。标准输出。流可以是可读的、可写的或两者兼而有之。所有流都是EventEmitter的实例。要访问流模块，要使用的语法是：

const stream = require('stream'); 
1

流的类型

Node.js 中有四种基本的流类型：

• Writable：可以写入数据的流（例如，fs.createWriteStream()）。
• Readable：可以从中读取数据的流（例如fs.createReadStream()）。
• Duplex：既是Writable又是Readable 的流（例如，net.Socket）。
• Transform：Duplex可以在写入和读取数据时修改或转换数据的流（例如，zlib.createDeflate()）。

// stream-copy.js

import {
  createReadStream,
  createWriteStream
} from 'fs'

const [,, src, dest] = process.argv

// 创建源流
const srcStream = createReadStream(src)

// 创建目标流
const destStream = createWriteStream(dest)

// 当源流上有数据时，
// 将其写入目标流
srcStream.on('data', (chunk) => destStream.write(chunk))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

本质上，我们用createReadStream和createWriteStream替换readFile和writeFile。然后使用它们创建两个流实例srcStream和destStream。这些对象分别是一个 ReadableStream（输入）和一个 WritableStream（输出）的实例。

目前，唯一需要理解的重要细节是流并不急切；他们不会一次性读取所有数据。数据以块、小部分数据的形式读取。一旦块通过data事件可用，我们就可以立即使用它。当源流中有新的数据块可用时，我们立即将其写入目标流。这样，我们就不必将所有文件内容保存在内存中。

请记住，这里的实现并不是万无一失的，存在一些粗糙的边缘情况，但就目前而言，这足以理解 Node.js 中流处理的基本原理。

可读流 → 该流用于创建用于读取的数据流，例如读取大块文件。

例子：

const fs = require('fs');

const readableStream = fs.createReadStream('./article.md', {
    highWaterMark: 10
});

readableStream.on('readable', () => {
    process.stdout.write(`[${readableStream.read()}]`);
});

readableStream.on('end', () => {
    console.log('DONE');
});
1
2
3
4
5
6
7
8
9
10
11
12
13

可写流 → 这将创建要写入的数据流。例如：向文件中写入大量数据。

例子：

const fs = require('fs'); 
const file = fs.createWriteStream('file.txt'); 
for (let i = 0; i < 10000; i++) 
{ 
file.write('Hello world ' + i); 
}
file.end();
1
2
3
4
5
6
7

双工流 → 该流用于创建同时可读和可写的流。

例子：

const server = http.createServer((req, res) => {
    let body = '';
    req.setEncoding('utf8');
    req.on('data', (chunk) => {
        body += chunk;
    });
    req.on('end', () => {
        console.log(body);
        try {
            res.write('Hello World');
            res.end();
        } catch (er) {
            res.statusCode = 400;
            return res.end(`error: ${er.message}`);
        }
    });
});
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

流动与非流动

Node 中有两种类型的可读流：

• 流动流 —— 用于从系统传递数据并将该数据提供给程序的流。
• 非流动流 —— 不自动推送数据的非流动流。相反，非流动流将数据存储在缓冲区中并显式调用read方法来读取它。

内存/时间比较

让我们看看这两种实现（缓冲区和流式传输）在内存使用和执行时间方面的比较。

我们可以查看 Node.js 脚本在缓冲区中分配了多少数据的一种方法是调用process.memoryUsage().arrayBuffers方法。

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);
1
2
3
4
5
6
7
8
9
10
11
12
13
14