nodejs:从events到stream

February 22, 2022

events 文档：http://nodejs.cn/api/events.html
stream 文档：http://nodejs.cn/api/stream.html

events

作为 异步事件驱动架构 的 Node.js，提供的 API 都是基于 events 设计的
由于设计模式相似，故形式上也与 DOM 的 Events 接口类似（但没有冒泡之类的）

const EventEmitter = require("events");

class MyEmitter extends EventEmitter {}

const myEmitter = new MyEmitter();
myEmitter.on("eventName", (a, b) => {
    console.log("an event occurred!");
});
myEmitter.once("disposable", () => console.log("只会触发一次"));
myEmitter.on("eventName", function (a, b) {
    // this 被绑定到当前 EventEmitter 实例
    console.assert(this === myEmitter);
});
myEmitter.addListener(eventName, listener); // 浏览器类似的API也是提供的，作为on的别名
myEmitter.emit("eventName", "arg1", ["arg2"]); // 多参可选

event-loop

EventEmitter 按照注册的顺序同步地调用所有监听器。
这里需要复习一下 javascript 事件循环（event-loop）机制
https://zh.javascript.info/event-loop

在浏览器中：
...->宏任务(事件回调触发)->微任务(如果有)->渲染(DOM)->宏任务->微任务->渲染->...
Promise 是微任务(jobs)，setTimeout、浏览器事件回调是宏任务(task) 宏任务: setTimeout setInterval setImmediate requestAnimationFrame I/O UI 渲染微任务: Promise MutationObserver process.nextTick queueMicrotask

// Node.js和浏览器表现略有不同

// 以下均为异步
// 0毫秒后将函数放进宏任务队列
setTimeout(() => console.log("setTimeout"), 0);
// 立即将函数放进宏任务队列
setImmediate(() => console.log("setImmediate")); // IE, node.js可用 现代浏览器中undefined
// 立即将函数放进微任务队列
Promise.resolve().then(() => console.log("Promise.resolve"));
// async/await 写法
(async () => {
    await Promise.resolve();
    console.log("async/await");
})();
// 立即将函数放进微任务队列的最前面！！！
process.nextTick(() => console.log("process.nextTick")); // node.js

js 引擎先顺序执行完所有同步任务
检查微任务，若非空则直接执行
执行完所有微任务，进入下一个事件循环
检查宏任务，按照顺序执行，执行完进入 2

Promise.resolve()
    .then(() => console.log("1"))
    .then(() => console.log("2"));
Promise.resolve().then(() => console.log("3"));
// => 1 3 2

这里是先将 1,3 放入当前队列
按顺序执行 1,3，这里 1 还将 2 放入下一次队列
执行完 1,3，开始下一次队列
执行 2

适当运用事件回调机制可以更好的调配事件的运行

回到 nodejs，错误事件

需要注意

myEmitter.emit("error", new Error("whoops!")); // 没有为 'error' 事件注册监听器。抛出错误，打印堆栈跟踪，然后退出 Node.js 进程。

myEmitter.on("error", (err) => {
    console.error("whoops! there was an error");
}); // 这样不会退出进程了

使用 async 函数作为监听器会造成抛出异常情况下导致的未处理的拒绝，这里不详细讨论，详细说明参见 node.js 文档

stream

nodejs 中的 stream 分为：Readable Writeable Duplex Transform

自动化处理流：readableSrc.pipe(writeableDest)
其中，Readable Duplex 属于 readableSrc，Writeable Duplex Transform 属于 writeableDest

流事件

readableSrc.on("data", (chunk) => {
    writeableDest.write(chunk);
});

readableSrc.on("end", (chunk) => {
    writeableDest.end();
});

可见，流也是基于事件回调的一种机制，pipe 方法则是对事件回调的封装

构造流

// 摘自文档
const { Writable } = require("stream");

class MyWritable extends Writable {
    constructor({ highWaterMark, ...options }) {
        super({ highWaterMark });
        // ...
    }
}

使用简化的构造函数方法构造流

const { Readable, Writable, Duplex, Transform } = require("stream");

const controller = new AbortController();

Writable

const myWritable = new Writable({
    construct(callback) {
        // 初始化状态并加载资源...
    },
    write(chunk, encoding, callback) {
        // chunk一般是一个Buffer
        console.log(chunk.toString());
        callback(); // 表示数据写入完成了，如果出错则传入一个Error即可
    },
    writev(chunks, callback) {
        // ...
    },
    destroy() {
        // 释放资源...
    },
    signal: controller.signal, // controller.abort();
});

// 使用
process.stdin.pipe(myWritable);

Readable

let max = 10,
    cur = 0;
const myReadable = new Readable({
    read(size) {
        if (++cur <= max) this.push(cur.toString());
    },
    signal: controller.signal, // controller.abort();
});

// 也可以在外部调用 myReadable.push("test")

// 使用
myReadable.pipe(process.stdout); // => 12345678910

Duplex

const myDuplex = new Duplex({
    read(size) {
        // ...
    },
    write(chunk, encoding, callback) {
        // ...
    },
});

Transform

const { Transform } = require("stream");

const controller = new AbortController();

const myTransform = new Transform({
    transform(chunk, encoding, callback) {
        this.push(chunk.toString().toUpperCase());
        callback();
    },
});

// 使用
process.stdin.pipe(myTransform).pipe(process.stdout);
// abc (enter) => ABC