【Node】Buffer 与 Stream

2023-05-16

node 为什么会出现 Buffer 这个模块

在最初的时候，JavaScript 只运行在浏览器端，

对于处理 Unicode 编码的字符串很容易，但是对于处理二进制以及非 Unicode 编码的数据便无能为力。

不过对于 Server 端操作来说网络I/O 以及文件I/O 的处理是必须的，所以 Node 中便提供了 Buffer 类处理二进制的数据。

二进制缓冲区 Buffer

一个 Buffer 类似于一个整数数组，可以取下标，有length属性，有剪切复制操作等，很多API也类似数组，但Buffer的大小在被创建时确定，且无法调整。

Buffer 可以与 String 互相转化，还可以设置字符集编码。

Buffer 用来处理文件 I/O、网络 I/O传输的二进制数据，String 用来呈现。

在处理文件 I/O、网络 I/O传输的二进制数据时，应该尽量以 Buffer 形式直接传输，速度会得到很好的提升。

Buffer 是一个典型的 JavaScript 与 C++ 结合的模块，与性能有关的用 C++ 来实现，JavaScript 负责衔接和提供接口。

Buffer 所占的内存不是 V8 堆内存，是独立于 V8 堆内存之外的内存，通过 C++ 层面实现内存申请。

可以说真正的内存是 C++层面提供的，而JavaScript分配内存可以说是 JavaScript层面只是使用它。

小结：Buffer 所占用的内存不是通过 V8 分配的，属于 堆外内存。

为了高效使用申请来的内存，Node 采用了 slab分配机制。slab 具有如下3种形态：

full：完全分配状态
partial：部分分配状态
empty：未被分配状态

Node 以 8KB 为界限来区分 Buffer 是大对象还是小对象：

Buffer.poolSize = 8 * 1024;

Buffer 对象是在 JavaScript层面的，能被 V8 的垃圾回收标记回收。

但是其内部的 parent 指向的 SlowBuffer 对象却来自于 Node 自身的 C++ 中的定义，是 C++层面上的Buffer 对象，所用内存不在 V8 的堆中。

小结：

真正的内存是在 Node 的 C++ 层面提供的，JavaScript 层面只是使用它。

当进行小而频繁的 Buffer 操作时，采用 slab 的机制进行预先申请和事后分配，使得 JavaScript 到操作系统之间不必有过多的内存申请方面的系统调用。

对于大块的 Buffer 而言，则直接使用 C++ 层面提供的内存，无需细腻的分配操作。

由于 Buffer太过常见，Node 进程在启动时就已经加载了它，并将其放在全局对象（global）上。所以在使用 Buffer时，无需通过 require() 即可直接使用。

Buffer 字符编码

通过使用字符编码，可以实现 Buffer实例与 JavaScript字符串之间的相互转换。

ascii —— 仅适用于7位ASCII数据。此编码速度很快，如果设置则会剥离高位。
utf8 —— 多字节编码的 Unicode字符。许多网页和其他文档格式都是用UTF-8。
base64 —— Base64编码。当从字符串创建Buffer时，此编码也会正确地接受RFC 4648第5节中指定的“URL和文件名安全字母”。
binary —— 一种将 Buffer编码成单字节编码字符串的方法。
hex —— 将每个字节编码成两个十六进制的字符。

字符串与 Buffer类型互传

字符串转 Buffer：Buffer.from()

const buf = Buffer.from('Node.js 技术栈', 'UTF-8');
 
console.log(buf); // <Buffer 4e 6f 64 65 2e 6a 73 20 e6 8a 80 e6 9c af e6 a0 88>
console.log(buf.length); // 17

Buffer 转字符串：toString([encoding], [start], [end])

const buf = Buffer.from('Node.js 技术栈', 'UTF-8');
 
console.log(buf); // <Buffer 4e 6f 64 65 2e 6a 73 20 e6 8a 80 e6 9c af e6 a0 88>
console.log(buf.length); // 17
console.log(buf.toString('UTF-8', 0, 9)); // 'Node.js �'

运行查看，可以看到以上输出结果为 Node.js �, 出现了乱码。

转换过程中为什么出现乱码？

问题出在这里一个中文在UTF-8下占用3个字节，“技”这个字在 buf 中对应的字节为 8a80e6，而我们设定的范围为0~9，

因此只输出了8a，这个时候就会造成字符被截断，出现乱码的情况。要想完整出现“技”这个字，则应该截取到11位。

console.log(buf.toString('UTF-8', 0, 11)); // 'Node.js 技'

Buffer 的拼接

Buffer 在使用场景中，通常是以一段一段的方式传输。

var fs = require('fs');
 
var rs = fs.createReadStream('test.txt');
var data = '';
rs.on('data', function (chunk) {
  data += chunk;
});
rs.on('end', function () {
  console.log(data);
});

上面这段代码是一个demo，用于流读取的示范，data事件中获取的chunk对象即是 Buffer对象。

对于初学者而言，容易将 Buffer当做字符串来理解，所以在接受上面的示例时不会觉得有任何异常。

一旦输入流中有宽字节编码时，问题就会暴露出来。这里潜藏的问题在于下面这句代码：

data += chunk;

这句代码里隐藏了 toString() 操作，它等价于如下的代码：

data =  data.toString() + chunk.toString();

值得注意的是，

外国人的语境通常是指英文环境，在他们的场景下，这个 toString() 不会造成任何问题。但是对于宽字节的中文，却会形成问题。

上面已经对乱码做了解释，所以这里主要看下可读流还有一个设置编码的方法 setEncoding()，示例如下：

readable.setEncoding(encoding)

该方法的作用是让 data事件中传递的不再是一个 Buffer对象，而是编码后的字符串。

事实上，在调用 setEncoding()时，可读流对象在内部设置了一个 decoder对象。

每次 data事件都通过该 decoder对象进行Buffer到字符串的编码，然后传递给调用者。

所以设置编码后，data 不再收到原始的 Buffer对象，但是这无法解释为何设置编码后乱码问题被解决掉的原因。

继续分析 decoder的神奇之处，它是来自于 string_decoder 模块的StringDecoder的实例对象。

StringDecoder 在得到编码后，知道宽字节字符串在 UTF-8编码下是以3个字节的方式存储的，

所以会确保返回的字符串不会包含 Buffer末尾中的任何不完整的多字节字符，并且会将不完整的字符保存在内部的buffer中用于下次调用。

虽然 string_decoder 模块很奇妙，但是它也并非万能药，它目前只能处理 utf-8、base64和ucs-2/utf-16le这3种编码，

所以，通过 setEncoding() 的方式不可否认能解决大部分的乱码问题，但并不能从根本上解决问题。

为了从根本上解决问题，+= 的方式显然不行，那么正确的 Buffer拼接方法应该如下面展示的形式：

var chunks = [];
var size = 0;
res.on('data', function (chunk) {
  chunks.push(chunk);
  size += chunk.length;
});
res.on('end', function () {
  var buf = Buffer.concat(chunks, size);
  var str = iconv.decode(buf, 'utf8');
  console.log(str);
});

正确的拼接方式是用一个数组来存储接收到的所有 Buffer片段并记录下所有片段的总长度。

然后调用 Buffer.concat()方法生成一个合并的 Buffer对象。Buffer.concat() 方法封装了从小Buffer对象向大Buffer对象的复制过程。

Buffer 内存分配与性能优化

Buffer 是一个典型的JavaScript与C++结合的模块，与性能有关的用C++来实现，JavaScript 负责衔接和提供接口。

Buffer 所占的内存不是V8分配的，是独立于V8堆内存之外的内存，通过C++层面实现内存申请、JavaScript分配内存。

每当我们使用 Buffer.alloc(size) 请求一个 Buffer内存时，Buffer 会以8KB为界限来判断分配的是大对象还是小对象，

小对象存入剩余内存池，不够再申请一个8KB的内存池；大对象直接采用C++层面申请的内存。

因此，对于一个大尺寸对象，申请一个大内存比申请众多小内存池要快很多。

Buffer Vs Cache

缓冲（Buffer）

Buffer 是用于处理二进制流数据，将数据缓冲起来，它是临时性的。

对于流式数据，会采用缓冲区将数据临时存储起来，等缓冲到一定的大小之后再存入硬盘中。

视频播放器就是一个经典的例子，有时你会看到一个缓冲的图标，这意味着此时这一组缓冲区并未填满。

当数据到达填满缓冲区并且被处理之后，此时缓冲图标消失，便可以看到一些图像数据了。

缓存（Cache）

Cache 可以看做是一个中间层，它可以是永久性的将热点数据进行缓存，使得访问速度更快。

例如我们通过 Memory、Redis 等将数据从硬盘或其他第三方接口中请求过来进行缓存，目的就是将数据存于内存的缓存区中，

这样对同一个资源进行访问时，速度会更快，这也是性能优化的一个重要的点。

流 Stream

举个例子，对于每一个客户端的请求，fs.readFile 接口都会把整个文件都缓存到内存中去，然后才开始把数据吐给用户。

那么当文件体积很大，请求也较多（且特别是当请求来自慢速用户）的时候，服务器需要消耗很大的内存，导致性能低下。

然而，这个问题正是 stream 发挥所长的地方。

不使用流：
在这里插入图片描述
使用流：

数据从A端流向B端与从B端流向A端是不一样的，因此，流是有方向的。输入流与输出流是相对的。

stream 的分类：

Readable Streams
Writeable Streams
Transform Streams（转换流 - 解析数据）：在读写过程中可以修改和变换数据的读写流。
Duplex Streams（双工流 - 接收和转换数据）：既可以读也可以写，如TCP连接，Socket连接等。
Classic Streams：经典接口，最早出现在 Node.js v0.4中，后被重新拆分，所以现在已经不怎么使用了。

在node中，这些流中的数据就是 Buffer对象，可读、可写流会将数据存储到内部的缓存中，等待被消费；

Duplex 和 Transform 则是维护了两个相互独立的缓存用于读和写。在维持了合理高效的数据流的同时，也使得对于读和写可以独立进行而互不影响。

在node中，这四种流都是 EventEmitter实例，它们都有 close、error事件，

可读流具有监听数据到来的data事件等，可写流则具有监听数据已传给底层系统的finish事件等。

Duplex 和 Transform 都同时实现了 Readable 和 Writeable 的事件和接口。

参考 Node.js 源码，可以发现 request 对象和 response 对象其实都是继承自 Stream 的。

// request
req => IncomingMessage => Stream.Readable
// response
res => ServerResponse => OutgoingMessage => Stream

很明显，HTTP 连接中的 request 对象是可读流（Stream.Readable），而 response 对象是完整的可读可写流（Stream.Duplex）。

从上文得知，请求和响应都是继承自 Stream的，所以可以直接通过 pipe 方法进行组装。

// const app = http.createServer((req, res) => {})
 
http.request => http.ClientRequest => OutgoingMessage => Stream
// http.request 通过pipe得到的返回值和res是一样的。
http.response => 继承自 EventEmitter，而不是Writeable Stream

值得一提的是 writeable的 drain事件，这个事件表示缓存的数据被排空了。为什么有这个事件呢？

起因是调用可写流的 write 和可读流的 read 都会有一个缓存区用来缓存写 /读的数据，

缓存区是有大小的，一旦写的内容超过这个大小，write 方法就会返回 false，表示写入停止，

这时如果继续 read 完缓存区数据，缓存区被排空，就会触发 drain事件，可以这样来防止缓存区爆仓：

var rs = fs.createReadStream(src);
var ws = fs.createWriteStream(dst);
 
rs.on('data', function (chunk) {
    if (ws.write(chunk) === false) {
        rs.pause();
    }
});
 
rs.on('end', function () {
    ws.end();
});
 
ws.on('drain', function () {
    rs.resume();
});

注意，只有可读流才具有pipe能力，可写流作为目的地。

pipe不仅可以作为通道，还能很好地控制管道里的流，控制读和写的平衡，不让任一方过度操作。

另外，pipe可以监听可读流的data、end事件，这样就可以构建快速的响应：

// 一个文件下载的例子，使用回调函数的话需要等到服务器读取完文件才能向浏览器发送数据
var http = require('http') ;
var fs = require('fs') ;
var server = http.createServer(function (req, res) {
    fs.readFile(__dirname + '/data.txt', function (err, data) {
        res.end(data);
    }) ;
}) ;
server.listen(8888) ;
 
// 而采用流的方式，只要建立连接，就会接受到数据，不用等到服务器缓存完data.txt
var http = require('http')
var fs = require('fs')
var server = http.createServer(function (req, res) {
    var stream = fs.createReadStream(__dirname + '/data.txt')
    stream.pipe(res)
})
server.listen(8888)

因此，使用 pipe 即可解决上面的那个爆仓问题。

关于数据流积压还可以参考数据流中的积压问题

接下来，继续从源码角度来分析下我感兴趣的 pipe：

Readable.prototype.pipe = function(dest, options) {
  const src = this;
  // Start the flow if it hasn't been started already.
  if (!state.flowing) {
    debug('pipe resume');
    src.resume();
  }
  src.on('data', ondata);
  function ondata(chunk) {
    const ret = dest.write(chunk);
    if (ret === false) {
      ...
      if (!ondrain) {
        // When the dest drains, it reduces the awaitDrain counter
        // on the source.  This would be more elegant with a .once()
        // handler in flow(), but adding and removing repeatedly is
        // too slow.
        ondrain = pipeOnDrain(src);
        dest.on('drain', ondrain);
      }
      src.pause();
    }
  }
  ...
  return dest;
};
 
// 当可写入流 dest 耗尽时，它将会在可读流对象 source 上减少 awaitDrain 计数器
// 为了确保所有需要缓冲的写入都完成，即 state.awaitDrain === 0 和 src 可读流上的 data 事件存在，切换流到流动模式
function pipeOnDrain(src) {
  return function pipeOnDrainFunctionResult() {
    const state = src._readableState;
    debug('pipeOnDrain', state.awaitDrain);
    if (state.awaitDrain)
      state.awaitDrain--;
    if (state.awaitDrain === 0 && EE.listenerCount(src, 'data')) {
      state.flowing = true;
      flow(src);
    }
  };
}
 
// stream.read() 从内部缓冲拉取并返回数据。如果没有可读的数据，则返回 null。在可读流上 src 还有一个 readable 属性，如果可以安全地调用 readable.read()，则为 true
function flow(stream) {
  const state = stream._readableState;
  debug('flow', state.flowing);
  while (state.flowing && stream.read() !== null);
}

在 Stream 的原型上声明 pipe 方法，订阅 data事件，src 为可读流对象，dest 为可写流对象。

我们在使用 pipe 方法的时候也是监听的 data事件，一边读取数据一边写入数据。

看下 ondata() 方法里的几个核心实现：

dest.write(chunk)：接收 chunk写入数据，如果内部的缓冲小于创建流时配置的 highWaterMark，则返回true；否则返回 false时应该停止向流写入数据，直到 ‘drain’ 事件被触发。
src.pause()：可读流会停止 data事件，意味着此时暂停数据写入了。
之所以调用 src.pause() 是为了防止读入数据过快来不及写入。

什么时候知道来不及写入呢？要看 dest.write(chunk) 什么时候返回false，是根据创建流时传的 highWaterMark 属性，默认为 16384（16kb），对象模式的流默认为16。

上面提到在 data事件里，如果调用 dest.write(chunk)返回false，就会调用 src.pause() 停止数据流动，什么时候再开启呢？

当可写入流 dest 耗尽时，它将会在可读流对象 source 上减少 awaitDrain 计数器。

为了确保所有需要缓冲的写入都完成，即 state.awaitDrain === 0 和 src 可读流上的 data 事件存在，切换流到流动模式。src.resume()

当可读流中没有数据可供消费时，调用 onend 函数，执行 dest.end() 方法，表明已没有数据要被写入可写流，然后进行关闭（关闭可写流的 fd）。

Buffer vs Stream

空间效率：

Stream 能够让我们去做一些通过 buffer 不可能做的事情，比如，获取一个很大的文件。

时间效率：

Stream 可以让我们处理数据更快，因为它是一边读取一边处理的，处理的文件越大，Stream 与 Buffer的时间差就越明显。

Stream 为什么要使用二进制 Buffer

Stream 中流动的数据是 Buffer 类型，也就是二进制。

Stream 最初的设计目的就是为了优化 IO操作（网络I/O 和文件I/O），

对应的后端无论是文件I/O 还是网络I/O，其中包含的数据格式都是未知的，

有可能是字符串、音频、视频、网络包等等。即使就是字符串，它的编码格式也是未知的，可能是ASCII编码，也可能是utf-8编码。

对于这些未知的情况，还不如直接使用最通用的格式 —— 二进制。

注意：

Buffer 虽好也不要瞎用，Buffer 与 String 两者都可以存储字符串类型的数据，

但是，String 与 Buffer不同，在内存分配上面，String 直接使用 V8堆存储，不用经过 C++堆外分配内存，

并且 Google 也对 String 进行了优化，在实际的拼接测速对比中，String 比 Buffer快。

但是 Buffer 的出现是为了处理二进制以及其他非Unicode编码的数据，所以在处理非utf8数据的手需要使用到 Buffer来处理。

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

Node

Buffer

Stream