Node

Posted on Edited on

介绍

根据 Node.js 官网的定义：Node.js 是一个基于 Chrome V8 引擎的 JavaScript 运行环境。 Node.js 使用了一个事件驱动、非阻塞式 I/O 的模型，使其轻量又高效。

Node.js 是基于 Chrome V8 引擎构建的.由事件循环分发 I/O 任务,最终工作线程将任务丢进线程池中去执行,而事件循环只要等待执行结果就可以了.

异步 IO

重复调用 IO 操作,判断是否完成,称为轮询.包括read,select,poll,kqueue,event ports.
期望实现无须主动判断的非阻塞 I/O.

异步 IO 优点

  • 前端通过异步 IO 可以消除阻塞。
  • 请求耗时少，假如有两个请求 A 和 B，那么异步 IO 用时为：Max（A+B）。同步则为 A+B，请求越多差距越大。
  • IO 是昂贵的，分布式 IO 是更昂贵的。
  • Node.js 适用于 IO 密集型，而不适用于 CPU 密集型。
  • 并不是所有都用异步任务好，遵循一个公式： s= (Ws+Wp)/(Ws+Wp/p) Ws 表示同步任务，Wp 表示异步任务，p 表示处理器的数量。

异步 IO 实现

  • 应用程序先将 JS 代码经 V8 转换为机器码。
  • 通过 Node.js Bindings 层，向操作系统 Libuv 的事件队列中添加一个任务。
  • Libuv 将事件推送到线程池中执行。
  • 线程池执行完事件，返回数据给 Libuv。
  • Libuv 将返回结果通过 Node.js Bindings 返回给 V8。
  • V8 再将结果返回给应用程序。

libuv

多种异步 IO 实现的抽象封装层.
libuv 实现了 Node.js 中 Eventloop,主要分为以下几个阶段:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
   ┌───────────────────────────┐
┌─>│           timers          │
│  └─────────────┬─────────────┘
│  ┌─────────────┴─────────────┐
│  │     pending callbacks     │
│  └─────────────┬─────────────┘
│  ┌─────────────┴─────────────┐
│  │       idle, prepare       │
│  └─────────────┬─────────────┘      ┌───────────────┐
│  ┌─────────────┴─────────────┐      │   incoming:   │
│  │           poll            │<─────┤  connections, │
│  └─────────────┬─────────────┘      │   data, etc.  │
│  ┌─────────────┴─────────────┐      └───────────────┘
│  │           check           │
│  └─────────────┬─────────────┘
│  ┌─────────────┴─────────────┐
└──┤      close callbacks      │
   └───────────────────────────┘

上图中每一个阶段都有一个先进先出的回调队列，只有当队列内的事件执行完成之后，才会进入下一个阶段。

  • timers：执行 setTimeout 和 setInterval 中到期的 callback。
  • pending callbacks：上一轮循环中有少数的 I/O callback 会被延迟到这一轮的这一阶段执行。
    • 执行一些系统操作的回调，例如 tcp 连接发生错误。
  • idle, prepare：仅内部使用。
  • poll：最为重要的阶段，执行 I/O callback(node 异步 api 的回调，事件订阅回调等)，在适当的条件下会阻塞在这个阶段。
    • 如果 poll 队列不为空，直接执行队列内的事件，直到队列清空。
    • 如果 poll 队列为空。
      • 如果有设置 setImmediate，则直接进入 check 阶段。
      • 如果没有设置 setImmediate，则会检查是否有 timers 事件到期。
        • 如果有 timers 事件到期，则执行 timers 阶段。
        • 如果没有 timers 事件到期，则会阻塞在当前阶段，等待事件加入。
  • check：执行 setImmediate 的 callback。
  • close callbacks：执行 close 事件的 callback，例如 socket.on("close",func)。

除此之外，Node.js 提供了 process.nextTick(微任务，promise 也一样) 方法，在以上的任意阶段开始执行的时候都会触发。

  • Event Loop 是一个很重要的概念，指的是计算机系统的一种运行机制。
  • Libuv 在 Linux 下基于 Custom Threadpool 实现。
  • Libuv 在 Windows 下基于 IOCP 实现。

常见的异步 IO 使用方式

  • 使用 step，q，async 等异步控制库。
  • 使用 Promise 处理异步。
  • 使用 EventEmitter，实现“发布/订阅”模式处理异步。
  • Node.js 暂不支持协程，可使用 Generator 代替。
  • 终极解决方案：async、await。

事件驱动架构

commonjs 模块

用法

  • 引入模块：require(“./index.js”)
  • 导出模块：module.exports={…} 或者 exports.key={}
    • 注意：exports 为 module.exports 的引用，不可使用 exports 直接赋值，模块无法导出，eg：exports={}
  • 缓存：require 值会缓存，通过 require 引用文件时，会将文件执行一遍后，将结果通过浅克隆的方式，写入全局内存，后续 require 该路径，直接从内存获取，无需重新执行文件
  • 值拷贝：模块输出是值的拷贝，一但输出，模块内部变化后，无法影响之前的引用，而 ESModule 是引用拷贝。commonJS 运行时加载，ESModule 编译阶段引用
    • CommonJS 在引入时是加载整个模块，生成一个对象，然后再从这个生成的对象上读取方法和属性
    • ESModule 不是对象，而是通过 export 暴露出要输出的代码块，在 import 时使用静态命令的方法引用制定的输出代码块，并在 import 语句处执行这个要输出的代码，而不是直接加载整个模块

      面试题：为何直接使用 exports 导出 commonjs 模块，eg：exports = “hello”
      解析：commonjs 模块是通过 module.exports 导出模块，exports 为 module.exports 的引用，使用 exports 直接赋值导出模块，只会改变 exports 的引用不会导出 commonjs 模块。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// cjs正确导出用法
exports.key = "hello world";
module.exports = "hello world";
//错误导出
exports = "hello world"; //无法输出

//解析：
const obj = {
  key: {},
};
obj.key = "hello world"; //可改变obj
const key = obj.key;
key.key1 = "hello world"; //可改变obj
key = "hello world"; //无法改变obj，改变了key的引用

module 属性

  • 任意 js 文件就是一个模块.可以直接使用 module 属性.
  • id: 返回模块标识符,一般是一个绝对路径
  • filename: 返回绝对路径
  • loaded: 返回布尔值,表示模块是否加载完成
  • parent: 返回对象存放调用当前模块的模块
  • children: 返回数组,存放当前模块调用的其他模块
  • exports;返回当前模块需要暴露的内容
  • paths: 返回数组,存放不同目录下的 node_modules 位置

原理实现

使用 fs、vm、path 内置模块，以及函数包裹形式实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
const vm = require("vm");
const path = require("path");
const fs = require("fs");

/**
 * commonjs的require函数：引入module
 * @param {string} filename module的名称
 */
function customRequire(filename) {
  const pathToFile = path.resolve(__dirname, filename);
  const content = fs.readFileSync(pathToFile, "utf-8");
  //使用函数包裹模块，执行函数
  //注入形参：require、module、exports、__dirname、__filename
  const wrapper = [
    "(function(require, module, exports, __dirname, __filename){",
    "})",
  ];
  const wrappedContent = wrapper[0] + content + wrapper[1];
  const script = new vm.Script(wrappedContent, {
    filename: "index.js",
  });
  const module = {
    exports: {},
  };
  //转换为函数，类似eval，(funcion(require, module, exports){ xxx })
  const result = script.runInThisContext();
  //函数执行，引入模块，若内部有require继续递归
  //exports为module.exports的引用
  result(customRequire, module, module.exports);
  return module.exports;
}

global.customRequire = customRequire;

源码分析

  • 源码路径：/lib/internal/modules/cjs/
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
//loader.js
//require函数定义
1、Module.prototype.require：调用__load函数
2、Module.__load：_cache处理，调用load函数
3、Module.prototype.load函数：调用Module._extensions[extension](this, filename);
//不同的后缀通过定义不同的函数指定解析规则：以Module._extensions['.js']为例
4、Module._extensions['.js'] = function(module, filename) {
  //读取缓存或者通过readFileSync读取内容
  if (cached?.source) {
    content = cached.source;
    cached.source = undefined;
  } else {
    content = fs.readFileSync(filename, 'utf8');
  }
  //...
  //调用compile解析
  module._compile(content, filename);
}
5、Module.prototype._compile = function(content, filename){
  //生成包裹函数：warpSafe获取函数字符串并使用vm执行生成执行函数
  const compiledWrapper = wrapSafe(filename, content, this);
  //执行函数
  const exports = this.exports;
  const thisValue = exports;
  const module = this;
  if (inspectorWrapper) {
    result = inspectorWrapper(compiledWrapper, thisValue, exports,
                              require, module, filename, dirname);
  } else {
    //静态方法Reflect.apply(target, thisArgument, argumentsList)
    //通过指定的参数列表发起对目标(target)函数的调用
    result = ReflectApply(compiledWrapper, thisValue,
                          [exports, require, module, filename, dirname]);
  }
  return result;
}
6、function wrapSafe(filename, content, cjsModuleInstance) {
    /* 生成包裹函数字符：
    let wrap = function(script) {
       return Module.wrapper[0] + script + Module.wrapper[1];
    };
    const wrapper = [
      '(function (exports, require, module, __filename, __dirname) { ',
      '\n});',
    ]；*/
  	const wrapper = Module.wrap(content);
    //获取包裹函数
    return vm.runInThisContext(wrapper, {
      filename,
      lineOffset: 0,
      displayErrors: true,
      importModuleDynamically: async (specifier) => {
        const loader = asyncESM.ESMLoader;
        return loader.import(specifier, normalizeReferrerURL(filename));
      },
    });
}
1
2
3
4
5
6
7
8
9
10
const Module = require("module");
//后缀解析扩展：.test后缀同.js后缀
Module._extensions[".test"] = Module._extensions[".js"];

//切面编程：解析js模块前做打印处理
const prevFunc = Module._extensions[".js"];
Module._extensions[".js"] = function (...args) {
  console.log("load script");
  prevFunc.apply(prevFunc, args);
};

事件循环

  • 同步代码
  • process.nextTick(优先级高于 promise)和 promise.then() ，之后进入事件循环
  • timers 阶段：这个阶段执行 timer（setTimeout、setInterval）的回调
  • I/O callbacks 阶段 ：处理一些上一轮循环中的少数未执行的 I/O 回调
  • idle, prepare 阶段 ：仅 node 内部使用
  • poll 阶段 ：获取新的 I/O 事件, 适当的条件下 node 将阻塞在这里
  • check 阶段 ：执行 setImmediate() 的回调
  • close callbacks 阶段：执行 socket 的 close 事件回调

执行流程

  • 执行同步代码
  • 查看是否有满足条件的微任务,执行
  • 执行 timer 中满足条件的宏任务
  • timer 中宏任务全部执行完毕切换队列
  • 切换前会清空微任务
  • 之后按照 timer => poll => check 检查是否有任务执行

和浏览器 eventloop 区别:

浏览器执行宏任务就会把本次循环生成的微任务清空,而 node 是在切换队列前清空,所以要将 timer 中宏任务全部执行完毕,再执行微任务.
习题解析：

  • new Promise(()=>{//同步执行}).then(()=>{//异步执行})
  • async function test(){console.log() //同步 -> await test(0) //同步 -> console.log()//异步}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
//习题1:
// 顺序不确定，只有两个语句，执行环境有差异
// 场景1: setTimeout 0 最少1ms，未推入timers队列中，执行结果为：setImmediate、setTimeout
// 场景2: setTimeout 0 已推入timers队列中，执行结果为：setTimeout、setImmediate
setTimeout(() => {
  console.log("setTimeout");
}, 0);
setImmediate(() => {
  console.log("setImmediate");
});

//习题2: 都在回调函数中，内容确定
//首轮事件循环setTimeout1的timers清空，执行至check阶段，先输出setImmediate
//第二轮事件循环setTimeout2
//最终输出：setTimeout1、setImmediate、setTimeout2
setTimeout(() => {
  setTimeout(() => {
    console.log("setTimeout2");
  }, 0);
  setImmediate(() => {
    console.log("setImmediate");
  });
  console.log("setTimeout1");
}, 0);

//习题3: 混合题
setTimeout(() => {
  console.log("timeout");
}, 0);
setImmediate(() => {
  console.log("immediate");
  setImmediate(() => {
    console.log("immediate1");
  });
  new Promise((resolve) => {
    console.log(77);
    resolve();
  }).then(() => {
    console.log(88);
  });
  process.nextTick(function () {
    console.log("nextTick1");
  });
});
new Promise((resolve) => {
  console.log(7);
  resolve();
}).then(() => {
  console.log(8);
});
process.nextTick(function () {
  console.log("nextTick2");
});
console.log("start");

// 第一轮：7 start -  timeout | immediate 77 | 8 ｜ nextTick2
// 第二轮：7 start nextTick2  8 timeout immediate 77 - immediate1 ｜ 88 ｜ nextTick1
// 第三轮：7 start nextTick2  8 timeout immediate 77 nextTick1 88 immediate1

模块

全局对象

Node 中全局对象是global,如果直接在模块中获取 this 指向的是{}.
而模块最终会封装进自执行函数中,指向的是global.

  • 全局对象：global，下挂如下对象和函数，使用时无需模块引入
    • global
    • Buffer、process、console、queueMicrotask
    • setTimeout、clearTimeout、setInterval、setImmediate、clearInterval
  • 模块中使用注入变量：
    • __filename：当前文件名称带路径，eg：/Users/jian/workspace/cjs/index.js
    • __dirname：当前文件夹路径，eg：/Users/jian/workspace/cjs/
    • cjs：require, module, exports
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
//全局this指向global
//模块文件中this指向：module.exports的对象
/* 模块中才可以使用的变量：commonjs模块注入，非模块中使用报错
__dirname、__filename、exports、module、require */

//node命令行中
> console.log(__dirname)
// Uncaught ReferenceError: __dirname is not defined
> console.log(this)
/* <ref *1> Object [global] {
  global: [Circular *1],
  clearInterval: [Function: clearInterval],
  clearTimeout: [Function: clearTimeout],
  setInterval: [Function: setInterval],
  setTimeout: [Function: setTimeout] {
    [Symbol(nodejs.util.promisify.custom)]: [Getter]
  },
  queueMicrotask: [Function: queueMicrotask],
  clearImmediate: [Function: clearImmediate],
  setImmediate: [Function: setImmediate] {
    [Symbol(nodejs.util.promisify.custom)]: [Getter]
  }
}*/

//node模块中执行
console.log(this)
console.log(__filename)
console.log(__dirname)
/*
{}
/Users/jian/workspace/cjs/index.js
/Users/jian/workspace/cjs
*/

process

  • argv：返回一个数组，由命令行执行脚本时的各个参数组成。它的第一个成员总是 node，第二个成员是脚本文件名，其余成员是脚本文件的参数
  • env：返回一个对象，成员为当前 shell 的环境变量
  • stdin：标准输入流
  • stdout：标准输出流
  • cwd()：返回当前进程的工作目录
1
2
3
4
console.log(process.argv);
//[ '/usr/local/bin/node', '/Users/jian/workspace/cjs/index.js' ]
console.log(process.cwd()); ////Users/jian/workspace/cjs
process.stdout.write("Hello World!"); //Hello World!

底层依赖模块

  • V8 引擎：主要是 JS 语法的解析，有了它才能识别 JS 语法
  • libuv：C 语⾔实现的⼀个⾼性能异步⾮阻塞 IO 库，⽤来实现 node.js 的事件循环
  • http-parser/llhttp：底层处理 http 请求，处理报⽂，解析请求包等内容
  • openssl：处理加密算法，各种框架运⽤⼴泛，底层依赖，无需 js 实现
  • zlib：处理压缩等内容

Eventmitter

  • 大多数时候我们不会直接使用 EventEmitter，而是在对象中继承它，包括 fs、net、 http 在内的，只要是支持事件响应的核心模块都是 EventEmitter 的子类。
  • EventEmitter 会按照监听器注册的顺序同步地调用所有监听器，所以必须确保事件的排序正确，且避免竞态条件。
  • 常见 api
    • addListener(event, listener)添加监听器
    • removeListener(event, listener)移除监听器
    • removeAllListeners([event])移除所有监听器
    • on(event, listener)注册监听器
    • off(eventName, listener)移除监听器
    • once(event, listener)为指定事件注册一个单次监听器
    • emit(event, [arg1], [arg2], […])按监听器的顺序执行执行每个监听器，如果事件有注册监听返回 true，否则返回 false
  • 错误处理：当 EventEmitter 实例出错时，应该触发 ‘error’ 事件。 这些在 Node 中被视为特殊情况。如果没有为 ‘error’ 事件注册监听器，则当 ‘error’ 事件触发时，会抛出错误、打印堆栈跟踪、并退出 Node.js 进程。作为最佳，应该始终为 ‘error’ 事件注册监听器
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
const EventEmitter = require("events").EventEmitter;
var eventEmitter = new EventEmitter();
eventEmitter.on("event", function (a, b) {
  console.log(a, b, this, this === eventEmitter);
  // 打印: 普通函数，this指向eventEmitter实例
  //   a b MyEmitter {
  //     domain: null,
  //     _events: { event: [Function] },
  //     _eventsCount: 1,
  //     _maxListeners: undefined } true
});
eventEmitter.emit("event", "a", "b");

eventEmitter.on("event", (a, b) => {
  //箭头函数，this指向module.exports
  console.log(a, b, this, this === module.exports);
  // 打印: a b {} true
});
eventEmitter.emit("event", "a", "b");

事件模型设计（使用：发布订阅模式）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class EventEmitter {
  constructor(maxListeners) {
    this.events = {};
    this.maxListeners = maxListners || Infinity;
  }

  on(event, listener) {
    if (!this.events[event]) {
      this.events[event] = [];
    }
    if (
      this.maxListener != Infinity &&
      this.events[event].length > this.maxListeners
    ) {
      console.warn(`${event} has reached max listeners.`);
      return this;
    }
    this.events[event].push(listener);
    return this;
  }

  off(event, listener) {
    if (!listener) {
      this.events[event] = null; //无listener全部移除
    } else {
      this.events[event] = this.events[event].filter(
        (item) => item !== listener
      );
    }
    return this; //链式调用
  }

  once(event, listener) {
    const func = (...args) => {
      this.off(event, listener);
      listener.apply(this, args);
    };
    this.on(event, func);
  }

  emit(event, ...args) {
    const listeners = this.events[event];
    if (!listeners) {
      console.log("no listeners");
      return this;
    }
    listeners.forEach((cb) => {
      cb.apply(this, args);
    });
    return this;
  }
}

常用模块

  • fs：⽂件系统，能够读取写⼊当前安装系统环境中硬盘的数据
  • path：路径系统，能够处理路径之间的问题
  • crypto：加密相关模块，能够以标准的加密⽅式对我们的内容进⾏加解密
  • dns：处理 dns 相关内容，例如我们可以设置 dns 服务器等等
  • http: 设置⼀个 http 服务器，发送 http 请求，监听响应等等
  • readline: 读取 stdin 的⼀⾏内容，可以读取、增加、删除我们命令⾏中的内容
  • os：操作系统层⾯的⼀些 api，例如告诉你当前系统类型及⼀些参数
  • vm: ⼀个专⻔处理沙箱的虚拟机模块，底层主要来调⽤ v8 相关 api 进⾏代码解析
  • promisify：可以把异步函数回调形式改为 promise 形式。

fs

  • 文件常识
    • 权限位 mode：文件所有者/文件所属组/其他用户的 rwx，eg：默认 0o666，对应十进制 438
    • 标志位 flag：文件的操作方式，r、w、s 同步、+增加相反操作、x 排他方式
    • 文件描述符 fd：识别和追踪文件
  • 完整性读写文件操作
    • fs.readFile(filename,[encoding],[callback(error,data)]
    • fs.writeFile(filename,data,[options],callback)
      • options：可选，为对象，{encoding, mode, flag}
      • 默认编码为 utf8，模式为 0666（可读可写可操作），flag 为 ‘w’
    • fs.unlink(filename, callback)
  • 指定位置读写文件操作(高级文件操作)
    • fs.open(path,flags,[mode],callback)
    • fs.read(fd, buffer, offset, length, position, callback);
    • fs.write(fd, buffer, offset, length, position, callback);
    • fs.close(fd,callback)
  1. fd：文件描述符，需要先使用 open 打开，使用 fs.open 打开成功后返回的文件描述符；
  2. buffer：一个 Buffer 对象，v8 引擎分配的一段内存，要将内容读取到的 Buffer；
  3. offset：整数，向 Buffer 缓存区写入的初始位置，以字节为单位；
  4. length：整数，读取文件的长度；
  5. position：整数，读取文件初始位置；文件大小以字节为单位
  6. callback：回调函数，有三个参数 err（错误），bytesRead（实际读取的字节数），buffer（被写入的缓存区对象），读取执行完成后执行。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
//nodejs中最好使用绝对路径
const pathToFile = path.resolve(__dirname, "./text");
//err first, 异步
fs.readFile(pathTofile, "utf-8", function (err, result) {
  if (err) {
    console.log("e", e);
    return err;
  }
  console.log("result", result);
});
//同步文件
fs.readFileSync(pathToFile, "utf-8");

//耦合promise封装
function read(path) {
  return new Promise(function (resolve, reject) {
    fs.readFile(path, { flag: "r", encoding: "utf-8" }, function (err, data) {
      if (err) {
        // 失败执行的内容
        reject(err);
      } else {
        // 成功执行的内容
        resolve(data);
      }
    });
  });
}

//通用promise封装，高版本已内置该库
function promisify(func) {
  return function (...args) {
    return new Promise((resolve, reject) => {
      args.push(function (err, result) {
        if (err) return reject(err);
        return resolve(result);
      });
      return func.apply(func, args);
      /* 等价于
      fs.readFile(path, { flag: 'r', encoding: 'utf-8' }, function (err, data) {
      	if (err) {
        	// 失败执行的内容
        	reject(err)
      	} else {
        	// 成功执行的内容
        	resolve(data)
      	}
    	}) */
    });
  };
}
const creadFileAsync = promisify(fs.readFile);
creadFileAsync(pathToFile, "utf-8")
  .then((content) => {
    console.log(content);
  })
  .catch((e) => {
    console.log("e", e);
  });

大文件拷贝

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
// copy 方法
function copy(src, dest, size = 16 * 1024, callback) {
  // 打开源文件
  fs.open(src, 'r', (err, readFd) => {
    // 打开目标文件
    fs.open(dest, 'w', (err, writeFd) => {
      let buf = Buffer.alloc(size);
      let readed = 0; // 下次读取文件的位置
      let writed = 0; // 下次写入文件的位置

      (function next() {
        // 读取
        fs.read(readFd, buf, 0, size, readed, (err, bytesRead) => {
          readed += bytesRead;

          // 如果都不到内容关闭文件
          if (!bytesRead) fs.close(readFd, err => console.log('关闭源文件'));

          // 写入
          fs.write(writeFd, buf, 0, bytesRead, writed, (err, bytesWritten) => {
            // 如果没有内容了同步缓存，并关闭文件后执行回调
            if (!bytesWritten) {
              fs.fsync(writeFd, err => {
                fs.close(writeFd, err => return !err && callback());
              });
            }
            writed += bytesWritten;

            // 继续读取、写入
            next();
          });
        });
      })();
    });
  });
}

目录(文件夹)操作

  • fs.mkdir: 创建目录
  • fs.rmdir: 删除目录
  • fs.access: 判断文件或目录是否有操作权限
  • fs.stat: 获取目录及文件信息
  • fs.readdir: 读取目录中内容
  • fs.unlink: 删除指定文件

path

  • path.join([path1][, path2][, …])：连接路径，主要用途在于，会正确使用当前系统的路径分隔符
  • path.resolve([from …], to)：用于返回绝对路径，类似于多次 cd 处理
  • path.dirname(p)：返回文件夹
  • path.extname(p)：返回后缀名
  • path.basename(p)：返回路径中的最后一部分,也就是名称
  • path.parse(): 解析路径,返回数组
  • path.format(): 序列化路径,将上面的数组重组成路径.
  • path.normalize(): 规范化路径
  • path.isAbsolute(): 判断传入路径是否是绝对路径
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
const path = require("path");

// 连接路径：/test/test1/2slashes/1slash
console.log('join path : ' + path.join('/test', 'test1', '2slashes/1slash', 'tab', '..'));

// 转换为绝对路径：/Users/jian/workspace/cjs/main.js
const mainPath = path.resolve('main.js')
console.log('resolve : ' + mainPath);
console.log(path.resolve()) // 什么都不传,返回绝对路径,不包含文件名.
console.log(path.resolve('a','b')) //返回绝对路径拼接/a/b
console.log(path.resolve('a',/b')) // 返回`/b`
// resolve接收两部分参数,如果to的部分有路径(包含/),就返回'/xxx',
// 如果不含路径,就将当前目录的绝对路径拼接上

// 路径中文件的后缀名：.js
console.log('ext name : ' + path.extname(mainPath));

// 路径最后名字：main.js
console.log('ext name : ' + path.basename(mainPath));

http

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
var http = require("http");
var fs = require("fs");
var url = require("url");

// 创建服务器
http
  .createServer(function (request, response) {
    // 解析请求，包括文件名
    var pathname = url.parse(request.url).pathname;

    // 输出请求的文件名
    console.log("Request for " + pathname + " received.");

    // 从文件系统中读取请求的文件内容
    fs.readFile(pathname.substr(1), function (err, data) {
      if (err) {
        console.log(err);
        // HTTP 状态码: 404 : NOT FOUND
        // Content Type: text/html
        response.writeHead(404, { "Content-Type": "text/html" });
      } else {
        // HTTP 状态码: 200 : OK
        // Content Type: text/html
        response.writeHead(200, { "Content-Type": "text/html" });

        // 响应文件内容
        response.write(data.toString());
      }
      //  发送响应数据
      response.end();
    });
  })
  .listen(8080);

// 控制台会输出以下信息
console.log("Server running at http://127.0.0.1:8080/");

Buffer

Buffer 用于读取或操作二进制数据流，做为 Node.js API 的一部分使用时无需 require，用于操作网络协议、数据库、图片和文件 I/O 等一些需要大量二进制数据的场景。Buffer 在创建时大小已经被确定且是无法调整的.
Buffer 是单独创建的,不占用 V8 的内存空间.

基本使用

Buffer 的创建
在此之前我们需要去简单的学习依一下 Buffer 的两个 API

  • Buffer.from(): 根据现有的数据结构去创建 buffer 数据
  • Buffer.alloc(): 指定 buffer 数据的长度来创建 buffer 数据
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 创建一个长度为 10、且用零填充的 Buffer。
const buf1 = Buffer.alloc(10); // <Buffer 00 00 00 00 00 00 00 00 00 00>

// 创建一个长度为 10、且用 0x1 填充的 Buffer。
const buf2 = Buffer.alloc(10, 1); // <Buffer 01 01 01 01 01 01 01 01 01 01>

// 创建一个长度为 10、且未初始化的 Buffer。
// 这个方法比调用 Buffer.alloc() 更快，
// 但返回的 Buffer 实例可能包含旧数据，
// 因此需要使用 fill() 或 write() 重写。
const buf3 = Buffer.allocUnsafe(10);

// 创建一个包含 [0x1, 0x2, 0x3] 的 Buffer。
const buf4 = Buffer.from([1, 2, 3]); // <Buffer 01 02 03>

// 创建一个包含 UTF-8 字节 [0x74, 0xc3, 0xa9, 0x73, 0x74] 的 Buffer。
const buf5 = Buffer.from("tést"); // <Buffer 74 c3 a9 73 74>

// 创建一个包含 Latin-1 字节 [0x74, 0xe9, 0x73, 0x74] 的 Buffer。
const buf6 = Buffer.from("tést", "latin1"); // <Buffer 74 e9 73 74>

Buffer 的读写
关于 Buffer 的数据读取在官网上有很多方法：如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
buf.readBigInt64BE([offset]);
buf.readBigInt64LE([offset]);
buf.readBigUInt64BE([offset]);
buf.readBigUInt64LE([offset]);
buf.readDoubleBE([offset]);
buf.readDoubleLE([offset]);
buf.readFloatBE([offset]);
buf.readFloatLE([offset]);
buf.readInt8([offset]);
buf.readInt16BE([offset]);
buf.readInt16LE([offset]);
buf.readInt32BE([offset]);
buf.readInt32LE([offset]);
buf.readIntBE(offset, byteLength);
buf.readIntLE(offset, byteLength);
buf.readUInt8([offset]);
buf.readUInt16BE([offset]);
buf.readUInt16LE([offset]);
buf.readUInt32BE([offset]);
buf.readUInt32LE([offset]);
buf.readUIntBE(offset, byteLength);
buf.readUIntLE(offset, byteLength);

这些方法名称有 BE 和 LE 的区别，这个大段和小段的意思，说白了就是方向的问题，而 Int8 是没有 BE 和 LE 的区别的，我们用代码来演示一下

1
2
3
4
5
6
7
8
buffer2.writeInt8(12, 1); // 在buffer2的第二位写入一个int8 的数字12
console.log(buffer2); // <Buffer 01 0c 03 04>

buffer2.writeInt16BE(512, 2); // 在buffer2的第三位开始写一个 int16的数字 512
console.log(buffer2); // <Buffer 01 0c 02 00>

buffer2.writeInt16LE(512, 2); // 在buffer2的第三位开始写一个 int16的数字 512
console.log(buffer2); // <Buffer 01 0c 00 02>

所以你看到了，因为 int8 类型的数字只占一位，所以不存在方向不方向的问题，而 int16 类型的数字要占两位，比如 512 用 16 进制表示就是 [02 00],所以使用 BE 表示正方向，而 LE 表示反方向。具体使用什么样的表示方法要和被人协商议定

实例属性

  • buf.length:返回内存中分配给 buf 的字节数。 不一定反映 buf 中可用数据的字节量。
  • buf.toString:根据 encoding 指定的字符编码将 buf 解码成字符串
  • buf.fill:用指定的 value 填充 buf。 如果没有指定 offset 与 end，则填充整个 buf
  • buf.equals:如果 buf 与参数具有完全相同的字节，则返回 true，否则返回 false
  • buf.indexOf:是否包含参数，包含则返回所在下标，不包含返回-1
  • buf.slice: 截取 buffer
  • buf.copy: 拷贝 buffer 中的数据

常用静态属性

Buffer.byteLength

这个表示整个 buffer 实际所占的字节数，因为不同的语言可能有不同的问题，比如英文字母在表达的时候一个字母就是一个字节，而中文用三个字节来表达汉字

1
2
3
4
5
console.log(Buffer.byteLength("test")); // 4
console.log(Buffer.byteLength("测试")); // 6

console.log(Buffer.from("test").length); // 4
console.log(Buffer.from("测试").length); // 6

Buffer.isBuffer

1
2
3
console.log(Buffer.isBuffer({})); // false
console.log(Buffer.isBuffer(Buffer.from("test"))); // true
console.log(Buffer.isBuffer(Buffer.from([1, 2, 3]))); // true

Buffer.concat

拼接 Buffer，注意的是里面不是传入多个 Buffer 对象，而是一个 Buffer 的数组

1
2
3
4
5
6
7
8
const buf1 = Buffer.from("This ");
const buf2 = Buffer.from("is ");
const buf3 = Buffer.from("buffer ");
const buf4 = Buffer.from("test ");
const buf5 = Buffer.from("demo ");

const bufConcat = Buffer.concat([buf1, buf2, buf3, buf4, buf5]);
console.log(bufConcat.toString()); // This is buffer test demo

自定义 Buffer: split 拆分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
ArrayBuffer.prototype.split = function (sep) {
  let len = Buffer.from(sep).length;
  let ret = [];
  let start = 0;
  let offset = 0;

  while ((offset = this.indexOf(sep, start) !== -1)) {
    ret.push(this.slice(start, offset));
    start = offset + len;
  }
  ret.push(this.slice(start));
  return ret;
};

let buf = "吃馒头, 吃面条, 吃所有";
let bufArr = buf.split("吃");

乱码问题

乱码的出现因为是汉字是三个字节的，现在 11 个字节不是三个倍数，没有办法正确按照每 3 个字节解析汉字，我们的解决方法是：使用内置的 string_decoder（字符串解码器）来解析 Buffer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
const { StringDecoder } = require("string_decoder");
const decoder = new StringDecoder("utf8");

const buf1 = Buffer.from("我正在学习");
console.log(buf1); // <Buffer e6 88 91 e6 ad a3 e5 9c a8 e5 ad a6 e4 b9 a0>
for (let i = 0; i < buf1.length; i += 5) {
  const buf2 = Buffer.allocUnsafe(5);
  buf1.copy(buf2, 0, i);
  console.log(decoder.write(buf2));
}
// <Buffer e6 88 91 e6 ad a3 e5 9c a8 e5 ad a6 e4 b9 a0>
//我
//正在
//学习

应用场景

I/O 操作

1
2
3
4
5
6
const fs = require("fs");

const inputStream = fs.createReadStream("input.txt"); // 创建可读流
const outputStream = fs.createWriteStream("output.txt"); // 创建可写流

inputStream.pipe(outputStream); // 管道读写

在 Stream 中我们是不需要手动去创建自己的缓冲区，在 Node.js 的流中将会自动创建。

加密解密

在一些加解密算法中会遇到使用 Buffer，例如 crypto.createCipheriv 的第二个参数 key 为 String 或 Buffer 类型，如果是 Buffer 类型，就用到了本篇我们讲解的内容，以下做了一个简单的加密示例，重点使用了 Buffer.alloc()初始化一个实例（这个上面有介绍），之后使用了 fill 方法做了填充，这里重点在看下这个方法的使用。 buf.fill(value[, offset[, end]][, encoding])

  • value: 第一个参数为要填充的内容
  • offset: 偏移量，填充的起始位置
  • end: 结束填充 buf 的偏移量
  • encoding: 编码集

扩展

Buffer 和 Cache 的区别

  • 缓冲（Buffer）是用于处理二进制流数据，将数据缓冲起来，它是临时性的，对于流式数据，会采用缓冲区将数据临时存储起来，等缓冲到一定的大小之后在存入硬盘中。视频播放器就是一个经典的例子，有时你会看到一个缓冲的图标，这意味着此时这一组缓冲区并未填满，当数据到达填满缓冲区并且被处理之后，此时缓冲图标消失，你可以看到一些图像数据。
  • 缓存（Cache）我们可以看作是一个中间层，它可以是永久性的将热点数据进行缓存，使得访问速度更快，例如我们通过 Memory、Redis 等将数据从硬盘或其它第三方接口中请求过来进行缓存，目的就是将数据存于内存的缓存区中，这样对同一个资源进行访问，速度会更快，也是性能优化一个重要的点。

Buffer 与 String 比较
在 HTTP 传输中传输的是二进制数据，那么如果使用 Buffer 和 String 情况如下：

  • 接口如果直接返回的字符串，这时候 HTTP 在传输之前会先将字符串转换为 Buffer 类型，以二进制数据传输，通过流（Stream）的方式一点点返回到客户端。
  • 但是直接返回 Buffer 类型，则少了每次的转换操作，对于性能也是有提升的。

所以在一些 Web 应用中，对于静态数据可以预先转为 Buffer 进行传输，可以有效减少 CPU 的重复使用（重复的字符串转 Buffer 操作）

stream

介绍

在 node 当中 stream 是一种处理流数据的抽象接口，stream 模块提供了一系列实现流的 API,在 node 当中提供了很多关于流的对象，比如 http 服务器的请求和 process.stdout 标准输出都是流的实例，流是可读的，可写的，同时也可以是可读写的，所有的流都是 EventEmitter 的实例。

之所以用 stream ，是因为一次性读取、操作大文件，内存和网络是“吃不消”的，因此要让数据流动起来，一点一点的进行操作,这其实也符合算法中一个很重要的思想 —— 分而治之

流转和应用

  • data 事件: 用来监听 stream 数据的流入
  • end 事件：用来监听 stream 数据输入的完成
  • pipe 方法: 用来做数据流转

① stream 从哪里来-soucre
stream 的常见来源方式有三种：

  • 从控制台输入
  • http 请求中的 request
  • 读取文件

这里先说一下从控制台输入这种方式，看一段 process.stdin 的代码：

1
2
3
4
process.stdin.on("data", function (chunk) {
  console.log("stream by stdin", chunk); // stream by stdin <Buffer 6b 6f 61 6c 61 6b 6f 61 6c 61 0a>
  console.log("stream by stdin", chunk.toString()); // stream by stdin koalakoala
});

然后从控制台输入任何内容都会被 data 事件监听到，process.stdin 就是一个 stream 对象,data 是 stream 对象用来监听数据传入的一个自定义函数， (说明： stream 对象可以监听”data”,”end”,”opne”,”close”,”error”等事件。node.js 中监听自定义事件使用.on 方法，例如 process.stdin.on(‘data’,…), req.on(‘data’,…),通过这种方式，能很直观的监听到 stream 数据的传入和结束)
② 连接水桶的管道-pipe
从水桶管道流转图中可以看到，在 source 和 dest 之间有一个连接的管道 pipe,它的基本语法是 source.pipe(dest) ，source 和 dest 就是通过 pipe 连接，让数据从 source 流向了 dest。
③ stream 到哪里去-dest
stream 的常见输出方式有三种：

  • 输出控制台
  • http 请求中的 response
  • 写入文件

我们上面借助控制台输入输出来讲解了 stream 的流转过程，可是在实际应用场景中，http 请求和文件操作当中使用 stream 的频率异常的频繁。原因是这样：
http 请求和文件操作都属于 IO ，即 stream 主要的应用场景就是处理 IO ，这就又回到了 stream 的本质 —— 由于一次性 IO 操作过大，硬件开销太多，影响软件运行效率，因此将 IO 分批分段操作，让数据一点一点的流动起来，直到操作完成 。

http 中的 steam

get 请求

1
2
3
4
5
6
7
8
9
10
11
12
const http = require("http");
const server = http.createServer(function (req, res) {
  const method = req.method; // 获取请求方法
  if (method === "GET") {
    // get 请求
    const fileName = path.resolve(__dirname, "data.txt");
    let stream = fs.createReadStream(fileName);
    stream.pipe(res); // 将 res 作为 stream 的 dest
  }
  // 其他method暂时忽略
});
server.listen(8000);

从上面例子可以看出，对 response 使用 stream 特性能提高性能。因此，在 nodejs 中如果要返回的数据是经过 IO 操作得来的，例如上面例子中读取文件内容，可以直接使用 stream.pipe(res) ,毕竟 response 也是一个 stream 对象，可以作为参数传入 pipe 当中，而不要再用 res.end(data)了。
这种应用的实例应该比较多，主要有两种场景：

  • 使用 node.js 作为服务代理，即客户端通过 node.js 服务作为跳板去请求其他服务，返回请求的内容
  • 使用 node.js 做静态文件服务器，直接返回静态文件

post 请求

web server 接收 http 请求肯定是通过 request，而 request 接收数据的本质其实就是 stream。所以 看似是 request 接收数据，但是在服务端的角度来说，request 就是产生数据的 source,那么 soucre 类型的 stream 对象都能监听 data,end 事件。分别触发数据接收和数据接收完成的通知，所以代码可以如下进行改造：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
var http = require("http");
var fs = require("fs");
var path = require("path");

var server = http.createServer(function (req, res) {
  var method = req.method; // 获取请求方法
  if (method === "POST") {
    // 暂只关注 post 请求
    req.on("data", function (chunk) {
      // 接收到部分数据
      console.log("chunk", chunk.toString().length);
    });
    req.on("end", function () {
      // 接收数据完成
      console.log("end");
      res.end("OK");
    });
  }
  // 其他请求方法暂不关心
});
server.listen(8000);

总结一下，request 和 response 一样，本身也是一个 stream 对象，可以用 stream 的特性，那肯定也能提高性能。两者的区别就在于，request 是 source 类型的、是 stream 的源头，而 response 是 dest 类型的、是 stream 的目的地。
这里再多举个例子，比如我们现在要将 node.js 接收到的 post 请求的数据写入文件，一般的小伙伴会这样写：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
var http = require("http");
var fs = require("fs");
var path = require("path");

var server = http.createServer(function (req, res) {
  var method = req.method; // 获取请求方法
  if (method === "POST") {
    // 暂只关注 post 请求
    var dataStr = "";
    req.on("data", function (chunk) {
      // 接收到数据，先存储起来
      var chunkStr = chunk.toString();
      dataStr += chunkStr;
    });
    req.on("end", function () {
      // 接收数据完成，将数据写入文件
      var fileName = path.resolve(__dirname, "post.txt");
      fs.writeFile(fileName, dataStr);

      res.end("OK");
    });
  }
  // 其他请求方法暂不关心
});
server.listen(8000);

这种写法也是对的，但是还没有真正理解 stream，当我们学习了文件操作中的 stream，你可能就 hi 写出更简单的代码，比如说下面这种：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
var http = require("http");
var fs = require("fs");
var path = require("path");

var server = http.createServer(function (req, res) {
  var method = req.method; // 获取请求方法
  if (method === "POST") {
    // 暂只关注 post 请求
    var fileName = path.resolve(__dirname, "post.txt");
    var writeStream = fs.createWriteStream(fileName);
    req.pipe(writeStream);
    req.on("end", function () {
      // 接收数据完成
      res.end("OK");
    });
  }
  // 其他请求方法暂不关心
});
server.listen(8000);

和 get 请求使用 stream 的场景类似，post 请求使用 stream 的场景，主要是用于将接收的数据直接进行 IO 操作，例如：

  • 将接受的数据直接存储为文件
  • 将接受的数据直接 post 给其他的 web server

fs 中的 stream

用 stream 读写文件其实前面章节都有多处用到，这里在统一整理一下：

  • 可以使用 fs.createReadStream(fileName)来创建读取文件的 stream 对象
  • 可以使用 fs.createWriteStream(fileName)来创建写入文件的 stream 对象

读取文件的 stream 对象，对应的就是 source，即数据的来源。写入文件的 steram 对象对应的就是 dest ，即数据的目的地。下面我们分别用普通的读写和使用 stream 实现一个文件拷贝的功能，并通过监控工具 memeye 来监控 node.js 内存占用的情况

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
const fs = require("fs");
const path = require("path");
const memeye = require("memeye");
memeye();
function copy(num) {
  const fileName1 = path.resolve(__dirname, "data.txt"); // 42kb左右
  fs.readFile(fileName1, function (err, data) {
    if (err) {
      console.log("读取出错", err.message);
      return;
    }
    var dataStr = data.toString();

    var fileName2 = path.resolve(__dirname, "data-bak.txt");
    fs.writeFile(fileName2, dataStr, function (err) {
      if (err) {
        console.log("写入出错", err.message);
        return;
      }
      console.log(`拷贝第${num}次成功`);
    });
  });
}

setTimeout(() => {
  for (let i = 0; i < 100; i++) {
    copy(i);
  }
}, 5000);

启动文件，在浏览器打开 localhost:23333/,可以看到 heapUsed 在 13.26M 左右，heapTotal 在 30M 左右，rss 在 40M 左右

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
const fs = require("fs");
const path = require("path");
const memeye = require("memeye");
memeye();
function copy(num) {
  var fileName1 = path.resolve(__dirname, "data.txt"); // 42KB左右
  var fileName2 = path.resolve(__dirname, "data-bak.txt");
  // 读取文件的stream对象
  var readStream = fs.createReadStream(fileName1);
  // 写入文件的stream对象
  var writeStream = fs.createWriteStream(fileName2);
  // 通过pipe实现数据的流转
  readStream.pipe(writeStream);

  // 监听数据完成的情况
  readStream.on("end", function () {
    console.log("拷贝完成");
  });
}

setTimeout(() => {
  for (let i = 0; i < 100; i++) {
    copy(i);
  }
}, 5000);

启动文件，在浏览器打开 localhost:23333/,可以看到 heapUsed 在 6.4M 左右，heapTotal 在 9.23M 左右，rss 在 30M 左右

总结：所有执行文件操作的场景，都应该尝试使用 stream ，例如文件的读写、拷贝、压缩、解压、格式转换等。除非是体积很小的文件，而且读写次数很少，性能上被忽略。如果是体积很大或者读写次数很多的情况下，建议使用 stream 来优化性能

逐行读取 readline

用 stream 操作文件，会来带很大的性能提升。但是原生的 stream 却对“行”无能为力，它只是把文件当做一个数据流、简单粗暴的流动。很多文件格式都是分行的，例如 csv 文件、日志文件，以及其他一些自定义的文件格式。

node.js 提供了非常简单的按行读取的 API —— readline，它本质上也是 stream，只不过是以“行”作为数据流动的单位。本节将结合一个分析日志文件的案例，讲解 readline 的使用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
const fs = require("fs");
const path = require("path");
const readline = require("readline");

var fileName = path.resolve(__dirname, "data.txt");
var readStream = fs.createReadStream(fileName);
// 创建readline对象
var readlineObj = readline.createInterface({
  input: readStream,
});

readlineObj.on("line", (lineDate) => {
  console.log(lineDate);
  console.log("---- line ----");
});

readlineObj.on("close", () => {
  console.log("end");
});

对 readline 最常见的使用应该是日志的逐行分析了，比如我们要从一个 10 万行的日志中去找一个 2018-10-23 14:00 这一分钟，对 user.html 的访问次数，我们就应该这样做

1
2
3
4
5
6
7
8
9
10
11
12
13
let num = 0
// 逐行读取日志内容，如果有包含`2018-10-23 14:00`和`user.html`的字符串就表示
// 在这一分钟有人访问了user.html,我们就记录一次
readlineObj.on('line'，function(lineData){
    if(lineData.indexOf('2018-10-23 14:00')>=0 && lineData.indexOf('user.html') >= 0){
        num++
    }
})

// 监听读取完成
readline.on('close',()=> {
    console.log('num',num)
})

借用这个简单的例子来演示 readline 的使用以及使用场景，其实日常工作中情况会更加复杂，不过再复杂的场景选择 readline 做逐行分析一定是不会错的。方向选对了很重要，选择大于努力.

stream 的种类

readable stream

可读流是对提供数据的来源的一种抽象。
可读流的例子包括：客户端的 HTTP 响应、服务器的 HTTP 请求、fs 的读取流、zilb 流、crypto 流、TCP socket、子进程 stdout 与 stderr、process.stdin，因为这些可读流都实现了 stream.Readable 类定义的接口

  • readable 事件: 当流中存在可读取数据时触发
  • data 事件: 当流中数据块传给消费者时触发

writeable stream

可写流是对数据要被写入的目的地的一种抽象.
可写流内部可写入的数据只支持 Buffer 和字符串.
可写流的例子包括：客户端的 HTTP 响应、服务器的 HTTP 请求、fs 的读取流、zilb 流、crypto 流、TCP socket、子进程 stdout 与 stderr、process.stdin，因为这些可写流都实现了 stream.Writable 类定义的接口

  • pipe 事件: 可读流调用 pipe()方法时触发
  • unpipe 事件: 可读流调用 unpipe()方法时触发

duplex stream

1
2
3
4
5
var fs = require("fs");
var zlib = require("zlib");
var readStream = fs.createReadStream("./data.txt");
var writeStream = fs.createWriteStream("./data-bak.txt");
readStream.pipe(zlib.createGzip()).pipe(writeStream);

pipe 的严谨用法要遵循下面三个原则：

  • 调用 pipe 的对象必须是 readable stream 或者 duplex stream 这样的具有读取数据的功能的流对象
  • pipe 中的参数对象必须是 writeable stream 或者 duplex stream 这样的具有写入数据的功能的流对象
  • pipe 支持链式调用

transform stream

转换流（Transform）是一种 Duplex 流，但它的输出与输入是相关联的,而且一般只有触发了写入操作时才会进入_transform 方法中。 与 Duplex 流一样， Transform 流也同时实现了 Readable 和 Writable 接口。

写入流程

  1. 第一次调用 write 方法时是将数据直接写入到文件中
  2. 第二次开始 write 方法就是将数据写入至缓存中
  3. 生产速度 和消费速度是不一样的，一般情况下生产速度要比消费速度快很多
  4. 当 flag 为 false 之后并不意味着当前次的数据不能被写入了.但是我们应该告之数据的生产者，当前的消费速度已经跟不上生产速度了，所以这个时候，一般我们会将可读流的模块修改为暂停模式。
  5. 当数据生产者暂停之后，消费者会慢慢的消化它内部缓存中的数据，直到可以再次被执行写入操作
  6. 当缓冲区可以继续写入事件如何让生产者知道? 使用 drain 事件.

背压机制

因为流的读取操作中,读的速度大于写的速度,为了控制,防止内存溢出,GC 频繁调用.

stream 有什么弊端

  • 用 rs.pipe(ws) 的方式来写文件并不是把 rs 的内容 append 到 ws 后面，而是直接用 rs 的内容覆盖 ws 原有的内容
  • 已结束/关闭的流不能重复使用，必须重新创建数据流
  • pipe 方法返回的是目标数据流，如 a.pipe(b) 返回的是 b，因此监听事件的时候请注意你监听的对象是否正确
  • 如果你要监听多个数据流，同时你又使用了 pipe 方法来串联数据流的话，你就要写成：
1
2
3
4
5
6
7
8
9
10
11
12
data
  .on("end", function () {
    console.log("data end");
  })
  .pipe(a)
  .on("end", function () {
    console.log("a end");
  })
  .pipe(b)
  .on("end", function () {
    console.log("b end");
  });

promisify

1
2
3
4
5
import { readFile } from "fs";
import { promisify } from "uitl";

// 该方法可以将read变为返回promise形式的异步函数
const read = promisify(readFile);

链表

可理解为一种存储数据的结构.
链表是一组节点组成的集合，每个节点都使用一个对象的引用来指向它的后一个节点。指向另一节点的引用讲做链。

链表结构图

1
data1|next => data2|next => data3|next => Null

data 中保存着数据，next 保存着下一个链表的引用。上图中，我们说 data2 跟在 data1 后面，而不是说 data2 是链表中的第二个元素。上图，值得注意的是，我们将链表的尾元素指向了 null 节点，表示链接结束的位置。

有头节点的链表

由于链表的起始点的确定比较麻烦，因此很多链表的实现都会在链表的最前面添加一个特殊的节点，称为 头节点，表示链表的头部。进过改造，链表就成了如下的样子：

1
Header|next => data1|next => data2|next => data3|next => Null

插入节点

向链表中插入一个节点的效率很高，需要修改它前面的节点(前驱)，使其指向新加入的节点，而将新节点指向原来前驱节点指向的节点即可。下面我将用图片演示如何在 data2 节点 后面插入 data4 节点。

1
2
3
4
Header|next => data1|next => data2|next => data3|next => Null
                                  	|					∧
  																	∨					|
                                    data4 | next

删除节点

同样，从链表中删除一个节点，也很简单。只需将待删节点的前驱节点指向待删节点的，同时将待删节点指向 null，那么节点就删除成功了。下面我们用图片演示如何从链表中删除 data4 节点

1
2
3
Header|next => data1|next => data2|next => data3|next  data4|next   Null
                                  								|										∧
  																								∨	----------------- |

单向链表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class Node {
  constructor(element, next) {
    this.element = element
    this.next = next
  }
}
class LinkedList {
  constructor(head, size) {
    this.head = null
    this.size = 0
  }
	// 获取节点
  _getNode(index){
    if(index<0 || index >= this.size) {
      throw new Error('越界')
    }
    // 通过循环链表获取prevNode
    let currentNode = this.head
    for(let i=0; i < index; i++) {
      currentNode = currentNode.next
    }
    return currentNode
  }
  // 增
  add(index, element) {
    if(argument.length ===1){
      element = index
      index = this.size
    }
    // 判断边界
    if(index < 0 || index > this.size) {
      throw new Error('越界
    }

    // 判断位置
    if(index === 0) {
      let head = this.head // 保存原有head的指向
      this.head = new Node(element, head) // 现在的head指向newNode
    }else {
      // 如果不在第一个位置,获取上一个节点
      let prevNode = this._getNode(index-1)
      // 上一Node 的next指向新节点
      prevNode.next = new Node(element, prevNode.next)
    }
    // add后注意size变化
    this.size++
  }
  // 删
  remove(index){
    let rmNode = null
    if(index === 0) {
      //如果删的是第一个
      rmNode = this.head
      if(!rmNode) {
        return undefined
      }
      this.head = head.next
    } else {
      let prevNode = _getNode(index -1)
      rmNode = prevNode.next
      prevNode.next = rmNode.next
    }
    this.size--
    return rmNode
  }
  // 改
  set(index, element) {
    let node = _getNode(index)
    node.element = element
  }
  // 查
  get(index) {
    return this._getNode(index)
  }
  // 清空
  clear(){
    this.head = null
    this.size = 0
  }
}
const l1 = new LinkedList()
li.add('node1')

单向链表队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class Queue {
  constructor() {
    this.linkedList = new LinkedList();
  }
  // 进队列
  enqueue(data) {
    this.linkedList.add(data);
  }
  // 出队列
  dequeue() {
    return this.linkedList.remove(0);
  }
}

const q = new Queue();