Node.js之async_hooks

February 8, 2024

async_hooks 不稳定，文档中也不建议使用。不过该包下提供 AsyncLocalStorage 类是稳定的，
其中的稳定方法可以使用。AsyncLocalStorage 是基于 async_hooks 实现的高层接口，因此有必要做一些了解。

async_hooks

async_hooks 用于跟踪异步资源，即，为原本的异步操作加上一些生命周期钩子。
异步资源表示具有关联（异步）回调的对象。
该回调可能被多次调用，例如 net.createServer() 的 'connection' 事件；
也可能只调用一次，例如 fs.open()。
资源也可以在（异步）回调被调用之前关闭。AsyncHook 没有明确区分这些不同的情况，而是将它们表示为抽象的概念，即资源。

在 Worker 线程中运行的代码拥有独立的 async_hooks 接口，每个线程都会使用一组新的 asyncID。

回调在异步作用域（async scope）中执行（即：执行上下文 execution context）。

注意：回调当然必须是异步执行。注意 Javascript 本身提供的异步回调实际上非常有限，
例如 setTimeout setInterval queueMicrotask 和 Promise 构造函数等
除了这些与事件循环有关的回调（事件循环也是由运行环境提供的）
其它异步回调则是由特定运行环境提供的，例如浏览器和 Node.js 提供的异步 API

import async_hooks from "node:async_hooks";

// 返回当前执行（execution）上下文 ID。执行上下文往往就是某个异步回调
const eid = async_hooks.executionAsyncId(); // => 0

// 返回触发（trigger）当前异步回调的上下文 ID。即当前回调是由哪个回调触发的
const tid = async_hooks.triggerAsyncId(); // => 0

// 通过提供一系列生命周期钩子（都是可选的），创建 AsyncHook 实例
const asyncHook = async_hooks.createHook({
    init,
    before,
    after,
    destroy,
    promiseResolve,
});

// 还需要调用此方法启用
asyncHook.enable();

// 还允许停止监听
asyncHook.disable();

AsyncHook 可以捕获 Node.js 提供的异步回调（即所谓异步资源）和 Promise
因为 Promise 也是异步资源，所以钩子不应该是异步函数（即返回 Promise的函数，防止无限递归）

注：使用同步方法即可，例如 fs.writeFileSync()
注意 Node.js 的 console.log() 是异步操作（可以想象是 fs.write(process.stdout.fd, xxx, () => {})）可用 process.stdout.write(xxx) 或 fs.writeSync(process.stdout.fd, xxx) 代替

我们自己创造的异步资源，则需要通过 AsyncResource 包装才能被捕获

import async_hooks from "node:async_hooks";
import { writeFileSync } from "node:fs";
import { format } from "node:util";

function debug(...args) {
    writeFileSync("log.out", `${format(...args)}\n`, { flag: "a" });
}

const asyncHook = async_hooks.createHook({
    init: debug,
    before: debug,
    after: debug,
    destroy: debug,
    promiseResolve: debug,
});

asyncHook.enable();

方法名	描述
`init(asyncId: number, type: string, triggerAsyncId: number, resource: object)`	在对象构造期间调用。当此回调运行时，资源可能尚未完成构造。因此，asyncId 引用的资源的所有字段可能尚未填充。
`before(asyncId: number)`	在资源的回调函数调用之前调用。对于事件监听器（如 TCPWrap），可以调用多次；对于请求（如 FSReqCallback），将仅调用一次。
`after(asyncId: number)`	在资源的回调函数完成后调用
`destroy(asyncId: number)`	当销毁资源时调用
`promiseResolve(asyncId: number)`	仅针对 Promise 调用。发生在当传递给 Promise 构造函数的 `resolve()` 函数被调用时

详细描述参见文档，这里说明一下 init 钩子的 type 参数：

type 是一个描述异步资源的字符串，通常对应于资源构造函数的名称。
由 Node.js 本身创建的资源类型可能在任何 Node.js 版本中更改。有效值包括 TLSWRAP、TCPWRAP、TCPSERVERWRAP、GETADDRINFOREQWRAP、FSREQCALLBACK、Microtask 和 Timeout。
由 AsyncResource 创建的与 Node.js 本身无关的异步资源，AsyncResource 的构造函数可提供 type。
还存在 PROMISE 资源类型，用于跟踪 Promise 实例和由它们调度的异步操作。

AsyncResource

完整类型见 @types/node

class AsyncResource {
    constructor(type: string, triggerAsyncId?: number | AsyncResourceOptions);

    static bind<Func extends (this: ThisArg, ...args: any[]) => any, ThisArg>(
        fn: Func,
        type?: string,
        thisArg?: ThisArg,
    ): Func;

    bind<Func extends (...args: any[]) => any>(fn: Func): Func;

    runInAsyncScope<This, Result>(fn: (this: This, ...args: any[]) => Result, thisArg?: This, ...args: any[]): Result;

    emitDestroy(): this;

    asyncId(): number;

    triggerAsyncId(): number;
}

AsyncLocalStorage

下面只列出了稳定 API，完整类型见 @types/node

AsyncLocalStorage 下文简称为 als

class AsyncLocalStorage<T> {
    getStore(): T | undefined;

    run<R>(store: T, callback: () => R): R;
    run<R, TArgs extends any[]>(store: T, callback: (...args: TArgs) => R, ...args: TArgs): R;
}

使用 als 可以实现隐式传递上下文

// ref: https://docs.adonisjs.com/guides/async-local-storage
import { AsyncLocalStorage } from "node:async_hooks";
import { setTimeout } from "node:timers/promises";

interface User {
    id: number;
}

const storage = new AsyncLocalStorage<{ user: User }>();

class UserService {
    async get() {
        // als 消费者与上下文无关，即无需显式持有上下文
        // 传统方式中可能要求将上下文作为参数传递
        // 换言之，上下文由 storage.run 隐式传递
        const state = storage.getStore(); // similar to thread-local storage
        const { user } = state!;
        console.log(`The user id is ${user.id}`);
        return user;
    }
}

async function run(user: User) {
    const state = { user };

    return storage.run(state, async () => {
        await setTimeout(100); // executing async function
        const userService = new UserService();
        return await userService.get(); // executing async function
    });
}

run({ id: 1 });
run({ id: 2 });
run({ id: 3 });

观察源码可以发现，实现也不复杂。
原理就是将上下文存储到资源中，每当新建资源时，让上下文从父资源传递到当前新建的资源。

const storageList: Array<AsyncLocalStorage> = [];

const storageHook = createHook({
    init(asyncId, type, triggerAsyncId, resource: object) {
        // init 钩子从参数中得到当前（子）资源
        const currentResource: object = executionAsyncResource(); // 获取父资源
        // Value of currentResource is always a non null object
        for (let i = 0; i < storageList.length; ++i) {
            // storageList 数组引用了所有 als 实例
            // 这里只是触发每个实例的 _propagate 方法
            storageList[i]._propagate(resource, currentResource, type);
        }
    },
});

class AsyncLocalStorage {
    constructor() {
        // 为异步资源关联一个 store
        // 这个 store 就是所谓的上下文
        this.kResourceStore = Symbol("kResourceStore");
        this.enabled = false;
    }

    _enable() {
        // 懒启动，对于每个 als 实例，都保存在 storageList 一次
        if (!this.enabled) {
            this.enabled = true;
            ArrayPrototypePush(storageList, this);
            storageHook.enable();
        }
    }

    /** 将父资源的 store 传递到 子资源的 store */
    _propagate(resource: object, triggerResource: object, type: string) {
        const store = triggerResource[this.kResourceStore];
        if (this.enabled) {
            resource[this.kResourceStore] = store;
        }
    }

    run(store, callback, ...args) {
        // Avoid creation of an AsyncResource if store is already active
        if (ObjectIs(store, this.getStore())) {
            return ReflectApply(callback, null, args);
        }

        this._enable();

        const resource = executionAsyncResource();
        const oldStore = resource[this.kResourceStore];

        resource[this.kResourceStore] = store;

        try {
            // run 方法运行的回调，回调中获取的 store 就是 run 方法传递的 store
            return ReflectApply(callback, null, args);
        } finally {
            // 还原 store
            resource[this.kResourceStore] = oldStore;
        }
    }

    getStore() {
        if (this.enabled) {
            const resource: object = executionAsyncResource();
            return resource[this.kResourceStore]; // 从当前资源中获得当前 store
        }
    }
}

←Back to home