Node.js精进(3)——流

虚幻大学 xuhss 316℃ 0评论

? 优质资源分享 ?

学习路线指引(点击解锁) 知识定位 人群定位
? Python实战微信订餐小程序 ? 进阶级 本课程是python flask+微信小程序的完美结合,从项目搭建到腾讯云部署上线,打造一个全栈订餐系统。
?Python量化交易实战? 入门级 手把手带你打造一个易扩展、更安全、效率更高的量化交易系统

  在 JavaScript 中,一般只处理字符串层面的数据,但是在 Node.js 中,需要处理网络、文件等二进制数据。

  由此,引入了BufferStream的概念,两者都是字节层面的操作。

  Buffer 表示一块专门存放二进制数据的缓冲区。Stream 表示流,一种有序、有起点和终点的二进制传输手段。

  Stream 会从 Buffer 中读取数据,像水在管道中流动那样转移数据。

  本系列所有的示例源码都已上传至Github,点击此处获取。

一、Buffer

  Buffer 是 JavaScript 中的 Uint8Array 的子类,Uint8Array 是一种类型化数组,处理 8 位无符号整数。

  其行为类似于数组(有 length 属性,可迭代等),但并不是真正的数组,其元素是 16 进制的两位数。

  Buffer 在创建时就会确定占用内存的大小,之后就无法再调整,并且它会被分配一块 V8 堆栈外的原始内存。

  Buffer 的应用场景比较多,例如在zlib模块中,利用 Buffer 来操作二进制数据实现资源压缩的功能;在crypto模块的一些加密算法,也会使用 Buffer。

1)创建

  在 Node 版本 <= 6 时,创建 Buffer 实例是 通过构造函数创建的:new Buffer(),但后面的版本就废弃了。

  现在常用的创建方法有:

  • Buffer.from() :传入已有数据,转换成一个 Buffer 实例,数据可以是字符串、对象、数组等。
  • Buffer.alloc():分配指定字节数量的 Buffer 实例。
  • Buffer.allocUnsafe() :功能与 Buffer.alloc() 相同,但其所占内存中的旧数据不会被清除,可能会泄漏敏感数据。

2)编码

  在创建一个 Buffer 实例后,就可以像数组那样访问某个字符,而打印出的值是数字,如下所示,这些数字是 Unicode 码。

let buf = Buffer.from('strick')
console.log(buf[0]);    // 115
console.log(buf[1]);    // 116

  若在创建时包含中文字符,那么就会多 3 个 16 进制的两位数,如下所示。

let buf = Buffer.from('strick')
console.log(buf); // 
buf = Buffer.from('strick平')
console.log(buf); // 

  Buffer.from() 的第二个参数是编码,默认值是 utf8,而 1 个中文字符经过 UTF-8 编码后通常会占用 3 个字节,1 个英文字符只占用 1 个字节。

  在调用 toString() 方法后就能根据指定编码(不传默认是 UTF-8)将 Buffer 解码为字符串。

console.log(buf.toString());    // strick平

  Node.js 支持的其他编码包括 latin1、base64、ascii 等,具体可参考官方文档

3)内存分配原理

  Node.js 内存分配都是在 C++ 层面完成的,采用 Slab 分配器(Linux 中有广泛应用)动态分配内存,并且以 8KB 为界限来区分是小对象还是大对象(参考自深入浅出Node.js)。

  可以简单看下Buffer.from()的源码,当它的参数是字符串时,其内部会调用 fromStringFast() 函数(在src/lib/buffer.js中),然后根据字节长度分别处理。

  如果当前所占内存不够,那么就会调用 createPool() 扩容,通过调用 createUnsafeBuffer() 创建 Buffer,其中 FastBuffer 继承自 Uint8Array。

// 以 8KB 为界限
Buffer.poolSize = 8 * 1024;
// Buffer.from() 内会调用此函数
function fromStringFast(string, ops) {
 const length = ops.byteLength(string);
 // 长度大于 4KB(>>> 表示无符号右移 1 位)
  if (length >= (Buffer.poolSize >>> 1))
 return createFromString(string, ops.encodingVal);
 // 当前所占内存不够(poolOffset 记录已经使用的字节数)
  if (length > (poolSize - poolOffset))
 createPool();
 let b = new FastBuffer(allocPool, poolOffset, length);
 const actual = ops.write(b, string, 0, length);
 if (actual !== length) {
 // byteLength() may overestimate. That's a rare case, though.
    b = new FastBuffer(allocPool, poolOffset, actual);
 }
 poolOffset += actual;
 alignPool();
 return b;
}
// 初始化一个 8 KB 的内存空间
function createPool() {
 poolSize = Buffer.poolSize;
 allocPool = createUnsafeBuffer(poolSize).buffer;
 markAsUntransferable(allocPool);
 poolOffset = 0;
}
// 创建 Buffer
function createUnsafeBuffer(size) {
 zeroFill[0] = 0;
 try {
 return new FastBuffer(size);
 } finally {
 zeroFill[0] = 1;
 }
}
// FastBuffer 继承自 Uint8Array
class FastBuffer extends Uint8Array {}

二、流

  流(Stream)的概念最早见于 Unix 系统,是一种已被证实有效的编程方式。

  Node.js 内置的流模块会被其他多个核心模块所依赖,它具有可读、可写或可读写的特点,并且所有的流都是 EventEmitter 的实例,也就是说被赋予了异步的能力。

  官方总结了流的两个优点,分别是:

  • 内存效率: 无需加载大量的数据到内存中即可进行处理。
  • 时间效率: 当获得数据之后就能立即开始处理数据,而不必等到整个数据加载完,这样消耗的时间就变少了。

1)流类型

  流的基本类型有4种:

  • Readable:只能读取数据的流,例如 fs.createReadStream(),可注册的事件包括 data、end、error、close等。
  • Writable:只能写入数据的流,例如 fs.createWriteStream(),HTTP 的请求和响应,可注册的事件包括 drain、error、finish、pipe 等。
  • Duplex:Readable 和 Writable 都支持的全双工流,例如 net.Socket,这种流会维持两个缓冲区,分别对应读取和写入,允许两边同时独立操作。
  • Transform:在写入和读取数据时修改或转换数据的 Duplex 流,例如 zlib.createDeflate()。

  来看一个官方的 Readable 流示例,先是用 fs.readFile() 直接将整个文件读到内存中。当文件很大或并发量很高时,将消耗大量的内存。

const http = require('http')
const fs = require('fs')

http.createServer(function(req, res) {
 fs.readFile(\_\_dirname + '/data.txt', (err, data) => {
 res.end(data)
 })
}).listen(1234)

  再用 fs.createReadStream() 方法通过流的方式来读取文件,其中 req 和 res 两个参数也是流对象。

  data.txt 文件中的内容将会一段段的传输给 HTTP 客户端,而不是等到读取完了再一次性响应,两者对比,高下立判。

http.createServer((req, res) => {
 const readable = fs.createReadStream(\_\_dirname + '/data.txt')
 readable.pipe(res);
}).listen(1234)

2)pipe()

  在上面的示例中,pipe() 方法的作用是将一个可读流 readable 变量中的数据传输到一个可写流 res 变量(也叫目标流)中。

  pipe() 方法地主要目的是平衡读取和写入的速度,让数据的流动达到一个可接受的水平,防止因为读写速度的差异,而导致内存被占满。

  在 pipe() 函数内部会监听可读流的 data 事件,并且会自动调用可写流的 end() 方法。

  当内部缓冲大于配置的最高水位线(highWaterMark)时,也就是读取速度大于写入速度时,为了避免产生背压问题,Node.js 就会停止数据流动。

  当再次重启流动时,会触发 drain 事件,其具体实现可参考此文

  pipe() 方法会返回目标流,虽然支持链式调用,但必须是 Duplex 或 Transform 流,否则会报错,如下所示。

http.createServer((req, res) => {
 const readable = fs.createReadStream(\_\_dirname + '/data.txt')
 const writable = fs.createWriteStream(\_\_dirname + '/tmp.txt')
 // Error [ERR\_STREAM\_CANNOT\_PIPE]: Cannot pipe, not readable
 readable.pipe(writable).pipe(res);
}).listen(1234)

3)end()

  很多时候写入流是不需要手动调用 end() 方法来关闭的。但如果在读取期间发生错误,那就不能关闭写入流,发生内存泄漏。

  为了防止这种情况发生,可监听可读流的错误事件,手动关闭,如下所示。

readable.on('error', function(err) {
 writeable.close();
});

  接下来看一种网络场景,改造一下之前的示例,让可读流监听 data、end 和 error 事件,当读取完毕或出现错误时关闭可写流。

http.createServer((req, res) => {
 const readable = fs.createReadStream(\_\_dirname + '/data.txt')
 readable.on('data', chunk => {
 res.write(chunk);
 });
 readable.on('end',() => {
 res.end();
 })
 readable.on('error', err => {
 res.end('File not found');
 });
}).listen(1234)

  若不手动关闭,那么页面将一直处于加载中,在KOA源码中,多处调用了此方法。

  注意,若取消对 data 事件的监听,那么页面也会一直处于加载中,因为流一开始是静止的,只有在注册 data 事件后才会开始活动。

4)大JSON文件

  网上看到的一道题,用 Node.js 处理一个很大的 JSON 文件,并且要读取到 JSON 文件的某个字段。

  直接用 fs.readFile() 或 require() 读取都会占用很大的内存,甚至超出电脑内存。

  直接用 fs.createReadStream() 也不行,读到的数据不能格式化成 JSON 对象,难以读取字段。

  CNode论坛上对此问题也做过专门的讨论

  借助开源库JSONStream可以实现要求,它基于jsonparse,这是一个流式 JSON 解析器。

  JSONStream 的源码去掉注释和空行差不多 200 行左右,在此就不展开分析了。

参考资料:

缓冲区 Stream多文件合并 pipe

legacy.js模块实现分析 Stream两种模式

Stream背压

深入理解Node.js之Buffer

Node.js Buffer Node.js 流

Node.js 语法基础 —— Buffter & Stream

node源码分析

通过源码解析 Node.js 中导流(pipe)的实现

转载请注明:xuhss » Node.js精进(3)——流

喜欢 (12)

您必须 登录 才能发表评论!