官方文档^[1]

async_hooks 模块提供了用于跟踪异步资源的 API。在最近的项目里用到了 Node.js 的 async_hooks 里的 AsyncLocalStorage Api。之前对 async_hooks 也有所耳闻，但没有实践过，正好趁此机会深入了解下。

什么是异步资源

这里的异步资源是指具有关联回调的对象，有以下特点：

1. 回调可以被一次或多次调用。比如 fs.open 创建一个 FSReqCallback 对象，监听 complete 事件，异步操作完成后执行一次回调，而 net.createServer 创建一个 TCP 对象，会监听 connection 事件，回调会被执行多次。
2. 资源可以在回调被调用之前关闭
3. AsyncHook 是对这些异步资源的抽象，不关心这些异步的不同
4. 如果用了 worker，每个线程会创建独立的 async_hooks 接口，使用独立的 asyncId。

为什么要追踪异步资源

因为 Node.js 基于事件循环的异步非阻塞 I/O 模型，发起一次异步调用，回调在之后的循环中才被调用，此时已经无法追踪到是谁发起了这个异步调用。

场景一

const fs = require('fs')
function callback(err, data) {
    console.log('callback', data)
}
fs.readFile("a.txt", callback)
console.log('after a')
fs.readFile("b.txt", callback)
console.log('after b')
// after a
// after b
// callback undefined
// callback undefined

我们用上面的例子代表 node 的异步 I/O，执行后的结果也符合我们的预期，那么问题来了，哪个 callback 是哪个 callback 呢？先执行的是 a 的还是 b 的呢？-> 我们无法从日志中确认调用链

场景二

function main() {
  setTimeout(() => {
    throw Error(1)
  }, 0)
}

main()
// Error: 1
//    at Timeout._onTimeout (/Users/zhangruiwu/Desktop/work/async_hooks-test/stack.js:3:11)
//    at listOnTimeout (internal/timers.js:554:17)
//    at processTimers (internal/timers.js:497:7)

异步回调抛出异常，也拿不到完整的调用栈。事件循环让异步调用和回调之间的联系断了。我：断了的弦，还怎么连？async_hooks: 听说有人找我 👀

AsyncHooks

下面是官方的一个 overview：

const async_hooks = require('async_hooks');
// 返回当前执行上下文的asyncId。
const eid = async_hooks.executionAsyncId();
// 返回触发当前执行上下文的asyncId。
const tid = async_hooks.triggerAsyncId();
// 创建asyncHook实例，注册各种回调
const asyncHook =
    async_hooks.createHook({ init, before, after, destroy, promiseResolve });
// 开启asyncHook，开启后才会执行回调
asyncHook.enable();
// 关闭asyncHook
asyncHook.disable();

//
// 下面是传入createHook的回调.
//

// 初始化异步操作时的钩子函数
function init(asyncId, type, triggerAsyncId, resource) { }

// 异步回调执行之前的钩子函数，可能触发多次
function before(asyncId) { }

// 异步回调完成后的钩子函数
function after(asyncId) { }

// 异步资源销毁时的钩子函数
function destroy(asyncId) { }

// 调用promiseResolve时的钩子函数
function promiseResolve(asyncId) { }

当开启 asyncHook 的时候，每个异步资源都会触发这些生命周期钩子。下面介绍了 init 的各个参数：

asyncId

异步资源的唯一 id，从 1 开始的自增

type

标识异步资源的字符串，下面是内置的一些 type，也可以自定义

FSEVENTWRAP, FSREQCALLBACK, GETADDRINFOREQWRAP, GETNAMEINFOREQWRAP, HTTPINCOMINGMESSAGE,

HTTPCLIENTREQUEST, JSSTREAM, PIPECONNECTWRAP, PIPEWRAP, PROCESSWRAP, QUERYWRAP,

SHUTDOWNWRAP, SIGNALWRAP, STATWATCHER, TCPCONNECTWRAP, TCPSERVERWRAP, TCPWRAP,

TTYWRAP, UDPSENDWRAP, UDPWRAP, WRITEWRAP, ZLIB, SSLCONNECTION, PBKDF2REQUEST,

RANDOMBYTESREQUEST, TLSWRAP, Microtask, Timeout, Immediate, TickObject

triggerAsyncId

触发当前异步资源初始化的异步资源的 asyncId。

const { fd } = process.stdout;

async_hooks.createHook({
  init(asyncId, type, triggerAsyncId) {
    const eid = async_hooks.executionAsyncId();
    fs.writeSync(
      fd,
      `${type}(${asyncId}): trigger: ${triggerAsyncId} execution: ${eid}\n`);
  }
}).enable();

net.createServer((conn) => {}).listen(8080);
// 启动后输出：
// TCPSERVERWRAP(4): trigger: 1 execution: 1 # 创建TCP Server，监听connect事件
// TickObject(5): trigger: 4 execution: 1    # listen 回调

// nc localhost 8080 后的输出：
// TCPWRAP(6): trigger: 4 execution: 0       # connect回调

新连接建立后，会创建 TCPWrap 实例，它是从 C++里被执行的，没有 js 堆栈，所以这里 executionAsyncId 是 0。但是这样就不知道是哪个异步资源导致它被创建，所以需要 triggerAsyncId 来声明哪个异步资源对它负责。

resource

一个代表异步资源的对象，可以从此对象中获得一些异步资源相关的数据。比如：GETADDRINFOREQWRAP 类型的异步资源对象，提供了 hostname。

使用示例

我们看一下官方的一个示例：

const { fd } = process.stdout;

let indent = 0;
async_hooks.createHook({
  init(asyncId, type, triggerAsyncId) {
    const eid = async_hooks.executionAsyncId();
    const indentStr = ' '.repeat(indent);
    fs.writeSync(
      fd,
      `${indentStr}${type}(${asyncId}):` +
      ` trigger: ${triggerAsyncId} execution: ${eid}\n`);
  },
  before(asyncId) {
    const indentStr = ' '.repeat(indent);
    fs.writeSync(fd, `${indentStr}before:  ${asyncId}\n`);
    indent += 2;
  },
  after(asyncId) {
    indent -= 2;
    const indentStr = ' '.repeat(indent);
    fs.writeSync(fd, `${indentStr}after:  ${asyncId}\n`);
  },
  destroy(asyncId) {
    const indentStr = ' '.repeat(indent);
    fs.writeSync(fd, `${indentStr}destroy:  ${asyncId}\n`);
  },
}).enable();

net.createServer().listen(8080, () => {
  // Let's wait 10ms before logging the server started.
  setTimeout(() => {
    console.log('>>>', async_hooks.executionAsyncId());
  }, 10);
});

启动服务后的输出：

TCPSERVERWRAP(4): trigger: 1 execution: 1  # listen 创建TCP server，监听connect事件
TickObject(5): trigger: 4 execution: 1     # 执行用户回调放在了nextTick里
before:  5
  Timeout(6): trigger: 5 execution: 5      # setTimeout
after:  5
destroy:  5
before:  6
>>> 6
  TickObject(7): trigger: 6 execution: 6   # console.log
after:  6
before:  7
after:  7

对第二行 TickObject 官方的解释是，没有 hostname 去绑定端口是个同步的操作，所以把用户回调放到nextTick^[2]去执行让它成为一个异步回调。所以一个思(mian)考(shi)题来了，求输出：

const net = require('net');
net.createServer().listen(8080, () => {console.log('listen')})

Promise.resolve().then(() => console.log('c'))
process.nextTick(() => { console.log('b') })
console.log('a')

TIPS：因为 console.log 是个异步操作，也会触发 AsyncHooks 回调。所以在 AsyncHooks 回调中执行 console 会无限循环：

const { createHook } = require('async_hooks');

createHook({
  init(asyncId, type, triggerAsyncId, resource) {
    console.log(222)
  }
}).enable()

console.log(111)
// internal/async_hooks.js:206
//     fatalError(e);
//     ^
//
// RangeError: Maximum call stack size exceeded
// (Use `node --trace-uncaught ...` to show where the exception was thrown)

可以用同步的方法输出到文件或者标准输出：

const { fd } = process.stdout // 1

createHook({
  init(asyncId, type, triggerAsyncId, resource) {
    // console.log(222)
    writeSync(fd, '222\n')
  }
}).enable()

console.log(111)

如何追踪异步资源

我们用 AsyncHooks 解决上面的两个场景的问题：

场景一

const fs = require('fs')
const async_hooks = require('async_hooks');
const { fd } = process.stdout;

let indent = 0;
async_hooks.createHook({
  init(asyncId, type, triggerAsyncId) {
    const eid = async_hooks.executionAsyncId();
    const indentStr = ' '.repeat(indent);
    fs.writeSync(
      fd,
      `${indentStr}${type}(${asyncId}):` +
      ` trigger: ${triggerAsyncId} execution: ${eid} \n`);
  },
  before(asyncId) {
    const indentStr = ' '.repeat(indent);
    fs.writeSync(fd, `${indentStr}before:  ${asyncId}\n`);
    indent += 2;
  },
  after(asyncId) {
    indent -= 2;
    const indentStr = ' '.repeat(indent);
    fs.writeSync(fd, `${indentStr}after:  ${asyncId}\n`);
  },
  destroy(asyncId) {
    const indentStr = ' '.repeat(indent);
    fs.writeSync(fd, `${indentStr}destroy:  ${asyncId}\n`);
  },
}).enable();

function callback(err, data) {
    console.log('callback', data)
}

fs.readFile("a.txt", callback)
console.log('after a')
fs.readFile("b.txt", callback)
console.log('after b')

FSREQCALLBACK(4): trigger: 1 execution: 1      # a
after a
TickObject(5): trigger: 1 execution: 1
FSREQCALLBACK(6): trigger: 1 execution: 1      # b
after b
before:  5
after:  5
before:  4
callback undefined
  TickObject(7): trigger: 4 execution: 4       # trigger by a
after:  4
before:  7
after:  7
before:  6
callback undefined
  TickObject(8): trigger: 6 execution: 6       # trigger by b
after:  6
before:  8
after:  8
destroy:  5
destroy:  7
destroy:  4
destroy:  8
destroy:  6

a 的调用链路：1 -> 4 -> 7b 的调用链路：1 -> 6 -> 8 所以第一个 callback 是 a，第二个 callback 是 b

场景二

const async_hooks = require('async_hooks');

function stackTrace() {
  const obj = {}
  Error.captureStackTrace(obj, stackTrace)
  return obj.stack
}

const asyncResourceMap = new Map();
async_hooks.createHook({
  init(asyncId, type, triggerAsyncId) {
    asyncResourceMap.set(asyncId, {
      asyncId,
      type,
      triggerAsyncId,
      stack: stackTrace()
    })
  },
  destroy(asyncId) {
    asyncResourceMap.delete(asyncId)
  },
}).enable();

function main() {
  setTimeout(() => {
    throw Error(1)
  }, 0)
}

main()

function getTrace(asyncId) {
  if (!asyncResourceMap.get(asyncId)) {
    return '';
  }
  const resource = asyncResourceMap.get(asyncId);
  if (resource?.triggerAsyncId) {
    getTrace(resource?.triggerAsyncId);
  }
  console.log(`${resource?.type}(${resource?.asyncId})\n${resource.stack}`)
}

process.on('uncaughtException', (err) => {
  console.log(getTrace(async_hooks.executionAsyncId()))
})

Timeout(2)
Error
    at AsyncHook.init (/Users/zhangruiwu/Desktop/work/async_hooks-test/async-error.js:16:14)
    at emitInitNative (internal/async_hooks.js:199:43)
    at emitInitScript (internal/async_hooks.js:467:3)
    at initAsyncResource (internal/timers.js:157:5)
    at new Timeout (internal/timers.js:191:3)
    at setTimeout (timers.js:157:19)
    at main (/Users/zhangruiwu/Desktop/work/async_hooks-test/async-error.js:25:3)
    at Object.<anonymous> (/Users/zhangruiwu/Desktop/work/async_hooks-test/async-error.js:30:1)
    at Module._compile (internal/modules/cjs/loader.js:1063:30)
    at Object.Module._extensions..js (internal/modules/cjs/loader.js:1092:10)

利用 AsyncHooks + Error.captureStackTrace 能追踪到完整的调用链

性能影响

使用 AsyncHook 会带来一定的性能开销：

https://github.com/bmeurer/async-hooks-performance-impact

// Node v14.16.0:
regular hapiserver: 11734.73 reqs.
init hapiserver: 8768.21 reqs.
full hapiserver: 6250.5 reqs.

regular koaserver: 17418.8 reqs.
init koaserver: 17183.6 reqs.
full koaserver: 14097.82 reqs.

实际应用

clinic 的性能测试工具Bubbleprof - Clinic.js^[3]就是利用 AsyncHooks + Error.captureStackTrace 追踪调用链。

AsyncResource

可自定义异步资源

const { AsyncResource, executionAsyncId } = require('async_hooks');

// 一般用于extend，实例化一个异步资源
const asyncResource = new AsyncResource(
  type, { triggerAsyncId: executionAsyncId(), requireManualDestroy: false }
);

// 在异步资源的执行上下文里运行函数:
// * 创建异步资源上下文
// * 触发before
// * 执行函数
// * 触发after
// * 恢复原始执行上下文
asyncResource.runInAsyncScope(fn, thisArg, ...args);

// 触发destory
asyncResource.emitDestroy();

// 返回asyncID
asyncResource.asyncId();

// 返回triggerAsyncId
asyncResource.triggerAsyncId();

下面是个例子

class MyResource extends asyncHooks.AsyncResource {
    constructor() {
        super('my-resource');
    }

    close() {
        this.emitDestroy();
    }
}

function p() {
    return new Promise(r => {
        setTimeout(() => {
            r()
        }, 1000)
    })
}

let resource = new MyResource;
resource.runInAsyncScope(async () => {
    console.log('hello')
    await p()
})

resource.close();

可以看到 runInAsyncScope 传入的函数是在我们自定义的异步资源回调里执行的:

my-resource(4): trigger: 1 execution: 1
before:  4
  PROMISE(5): trigger: 4 execution: 4
hello
  TickObject(6): trigger: 4 execution: 4
  PROMISE(7): trigger: 4 execution: 4
  Timeout(8): trigger: 4 execution: 4
  PROMISE(9): trigger: 7 execution: 4
after:  4
before:  6
after:  6
destroy:  4
destroy:  6
before:  8
after:  8
before:  9
after:  9
destroy:  8

AsyncLocalStorage

用于在回调和 Promise 链中创建异步状态。它允许在 Web 请求的整个生命周期或任何其他异步持续时间内存储数据。类似于其他语言中的线程本地存储(TLS)。

const http = require('http');
const { AsyncLocalStorage } = require('async_hooks');

const asyncLocalStorage = new AsyncLocalStorage();

function logWithId(msg) {
  const id = asyncLocalStorage.getStore();
  console.log(`${id !== undefined ? id : '-'}:`, msg);
}

let idSeq = 0;
http.createServer((req, res) => {
  asyncLocalStorage.run(idSeq++, () => {
    logWithId('start');
    // Imagine any chain of async operations here
    setImmediate(() => {
      logWithId('finish');
      res.end();
    });
  });
}).listen(8080);

http.get('http://localhost:8080');
http.get('http://localhost:8080');
// Prints:
//   0: start
//   1: start
//   0: finish
//   1: finish

实现原理

https://github.com/nodejs/node/blob/master/lib/async_hooks.js#L267

const storageList = [];
const storageHook = createHook({
  init(asyncId, type, triggerAsyncId, resource) {
    const currentResource = executionAsyncResource();
    // Value of currentResource is always a non null object
    for (let i = 0; i < storageList.length; ++i) {
      storageList[i]._propagate(resource, currentResource);
    }
  }
});

class AsyncLocalStorage {
  constructor() {
    this.kResourceStore = Symbol('kResourceStore');
    this.enabled = false;
  }

  disable() {
    if (this.enabled) {
      this.enabled = false;
      // If this.enabled, the instance must be in storageList
      storageList.splice(storageList.indexOf(this), 1);
      if (storageList.length === 0) {
        storageHook.disable();
      }
    }
  }

  // Propagate the context from a parent resource to a child one
  _propagate(resource, triggerResource) {
    const store = triggerResource[this.kResourceStore];
    if (this.enabled) {
      resource[this.kResourceStore] = store;
    }
  }

  enterWith(store) {
    if (!this.enabled) {
      this.enabled = true;
      storageList.push(this);
      storageHook.enable();
    }
    const resource = executionAsyncResource();
    resource[this.kResourceStore] = store;
  }

  run(store, callback, ...args) {
    const resource = new AsyncResource('AsyncLocalStorage');
    return resource.runInAsyncScope(() => {
      this.enterWith(store);
      return callback(...args);
    });
  }

  exit(callback, ...args) {
    if (!this.enabled) {
      return callback(...args);
    }
    this.enabled = false;
    try {
      return callback(...args);
    } finally {
      this.enabled = true;
    }
  }

  getStore() {
    const resource = executionAsyncResource();
    if (this.enabled) {
      return resource[this.kResourceStore];
    }
  }
}

run：

• 创建 AsyncResource，存储 store

• 开启 asyncHook(init)

• 执行 run 传入的函数

• 创建新的异步资源时触发 init 回调

• 将当前异步资源的 store 传递给新的异步资源

• 依次类推

getStore：从当前异步资源获取 store

性能影响

Kuzzle^[4]的性能基准测试，使用了 AsyncLocalStorage 与未使用之间相差 ～ 8%

let fn = async () => /test/.test('test');

Performed 180407 iterations to warmup
Performed 205741 iterations (with ALS enabled)
Performed 6446728 iterations (with ALS disabled)
ALS penalty: 96.8%

let fn = promisify(setTimeout).bind(null, 2);
Performed 44 iterations to warmup
Performed 4214 iterations (with ALS enabled)
Performed 4400 iterations (with ALS disabled)
ALS penalty: 4.23%

所以还是需要结合实际应用场景评估性能影响，als 带来的损耗跟业务代码相比可能微不足道。

应用场景

实现 CLS

Continuation-local storage\(CLS\)^[6]类似其他语言的线程本地存储(TLS), 得名于函数式编程中的延续传递风格Continuation-passing style^[7](CPS)，CPS 就类似 node 中链式回调的风格，旨在链式函数调用过程中维护一个持久的数据。cls-hooked^[8]是用 async_hooks 实现 CLS 的一个库，对不支持 async_hooks 的版本也做了兼容asynchronous-local-storage^[9]是用 ALS 实现的，对老版本会回退到 cls-hooked, 也是这次一体化项目里用到的。下面是官方的示例

const { als } = require('asynchronous-local-storage')
const express = require('express')
const app = express()
const port = 3000

app.use((req, res, next) => {
  als.runWith(() => {
    next();
    }, { user: { id: 'defaultUser' } }); // sets default values
});

app.use((req, res, next) => {
  // overrides default user value
  als.set('user', { id: 'customUser' });
  next();
});

app.get('/', (req, res) => res.send({ user: als.get('user') }))

app.listen(port, () => console.log(`Example app listening at http://localhost:${port}`))

用它我们可以在异步调用链上共享数据，想获取数据时随时随地调一下 get 即可。再看一下上面的例子，既然我们可以在异步调用链上共享数据了，那我们中间件还需要传递 req, res 参数吗？express 和 koa 的中间件设计都需要我们在每个中间件间传递固定的参数，想要获取一些就需要从中间件参数里拿到 ctx。ALS 可以打破这个限制，获取 ctx 不必从参数里拿，没有了这个限制我们可以做一些不一样的事情，比如 midway 的一体化调用方案^[10]。最近比较火的函数式框架farrow^[11]也是利用了 ALS 给我们带来了不一样的开发体验。

参考资料

• Async hooks | Node.js v16.1.0 Documentation^[12]

• bmeurer/async-hooks-performance-impact^[13]

• Making async_hooks fast \(enough\)^[14]

• NodeJs async_hooks 详解^[15]

• 在 Node.js 中使用 Async Hooks 处理 HTTP 请求上下文实现链路追踪^[16]

• https://medium.com/nmc-techblog/the-power-of-async-hooks-in-node-js-8a2a84238acb

• https://blog.kuzzle.io/nodejs-14-asynclocalstorage-asynchronous-calls

• https://itnext.io/request-id-tracing-in-node-js-applications-c517c7dab62d

参考资料