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) // <Buffer 43 69 61 6f 20 68 75 6d 61 6e>
console.log(bufferFromByteArray) // <Buffer 43 69 61 6f 20 68 75 6d 61 6e>
console.log(bufferFromHex) // <Buffer 43 69 61 6f 20 68 75 6d 61 6e>
console.log(bufferFromBase64) // <Buffer 43 69 61 6f 20 68 75 6d 61 6e>// 原始缓冲区数据可以“可视化”为字符串、十六进制或 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

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

// buffer-copy.jsimport {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)})

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

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

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

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

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

为什么会发生这种情况？

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

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

// biggest-buffer.jsimport buffer from 'buffer'// 这将分配几 GB 内存
const biggestBuffer = Buffer.alloc(buffer.constants.MAX_LENGTH) // 创建一个具有最大可能大小的缓冲区
console.log(biggestBuffer) // <Buffer 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ... 4294967245 more bytes>

在某种程度上，我们可以将流视为一种抽象，它允许我们处理在不同时刻到达的数据部分（块）。每个块都是一个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'); 

流的类型

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

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

// stream-copy.jsimport {createReadStream,createWriteStream
} from 'fs'const [,, src, dest] = process.argv// 创建源流
const srcStream = createReadStream(src)// 创建目标流
const destStream = createWriteStream(dest)// 当源流上有数据时，
// 将其写入目标流
srcStream.on('data', (chunk) => destStream.write(chunk))

本质上，我们用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');
});

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

例子：

const fs = require('fs'); 
const file = fs.createWriteStream('file.txt'); 
for (let i = 0; i < 10000; i++) 
{ 
file.write('Hello world ' + i); 
}
file.end();

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

例子：

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}`);}});
});

流动与非流动

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);