异步上下文跟踪

import { AsyncLocalStorage, AsyncResource } from 'node:async_hooks';

import http from 'node:http';
import { AsyncLocalStorage } from 'node: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');
    // 在此处想象任何异步操作链
    setImmediate(() => {
      logWithId('finish');
      res.end();
    });
  });
}).listen(8080);

http.get('http://localhost:8080');
http.get('http://localhost:8080');
// 打印：
//   0: start
//   0: finish
//   1: start
//   1: finish

import { AsyncLocalStorage } from 'node:async_hooks';

const asyncLocalStorage = new AsyncLocalStorage();

{
  using _ = asyncLocalStorage.withScope('my-store');
  console.log(asyncLocalStorage.getStore()); // 打印：my-store
}

console.log(asyncLocalStorage.getStore()); // 打印：undefined

import { AsyncLocalStorage } from 'node:async_hooks';

const asyncLocalStorage = new AsyncLocalStorage();

try {
  using _ = asyncLocalStorage.withScope('my-store');
  console.log(asyncLocalStorage.getStore()); // 打印：my-store
  throw new Error('test');
} catch (e) {
  // 即使出错后存储也会自动恢复
  console.log(asyncLocalStorage.getStore()); // 打印：undefined
}

import { AsyncLocalStorage } from 'node:async_hooks';

const storage = new AsyncLocalStorage();

// 如果不使用 using，必须手动处置作用域
const scope = storage.withScope('my-store');
// 此处 storage.getStore() === 'my-store'

scope.dispose(); // 恢复之前的值
// 此处 storage.getStore() === undefined

import { AsyncResource, executionAsyncId } from 'node:async_hooks';

// AsyncResource() 旨在被扩展。实例化一个
// 新的 AsyncResource() 也会触发 init。如果省略 triggerAsyncId，则
// 使用 async_hook.executionAsyncId()。
const asyncResource = new AsyncResource(
  type, { triggerAsyncId: executionAsyncId(), requireManualDestroy: false },
);

// 在资源的执行上下文中运行函数。这将
// * 建立资源的上下文
// * 触发 AsyncHooks before 回调
// * 使用提供的参数调用提供的函数 `fn`
// * 触发 AsyncHooks after 回调
// * 恢复原始执行上下文
asyncResource.runInAsyncScope(fn, thisArg, ...args);

// 调用 AsyncHooks destroy 回调。
asyncResource.emitDestroy();

// 返回分配给 AsyncResource 实例的唯一 ID。
asyncResource.asyncId();

// 返回 AsyncResource 实例的触发 ID。
asyncResource.triggerAsyncId();

import { parentPort } from 'node:worker_threads';
parentPort.on('message', (task) => {
  parentPort.postMessage(task.a + task.b);
});

import { AsyncResource } from 'node:async_hooks';
import { EventEmitter } from 'node:events';
import { Worker } from 'node:worker_threads';

const kTaskInfo = Symbol('kTaskInfo');
const kWorkerFreedEvent = Symbol('kWorkerFreedEvent');

class WorkerPoolTaskInfo extends AsyncResource {
  constructor(callback) {
    super('WorkerPoolTaskInfo');
    this.callback = callback;
  }

  done(err, result) {
    this.runInAsyncScope(this.callback, null, err, result);
    this.emitDestroy();  // `TaskInfo` 仅使用一次。
  }
}

export default class WorkerPool extends EventEmitter {
  constructor(numThreads) {
    super();
    this.numThreads = numThreads;
    this.workers = [];
    this.freeWorkers = [];
    this.tasks = [];

    for (let i = 0; i < numThreads; i++)
      this.addNewWorker();

    // 每当发出 kWorkerFreedEvent 时，调度
    // 队列中的下一个待处理任务（如果有）。
    this.on(kWorkerFreedEvent, () => {
      if (this.tasks.length > 0) {
        const { task, callback } = this.tasks.shift();
        this.runTask(task, callback);
      }
    });
  }

  addNewWorker() {
    const worker = new Worker(new URL('task_processor.js', import.meta.url));
    worker.on('message', (result) => {
      // 成功时：调用传递给 `runTask` 的回调，
      // 移除与 Worker 关联的 `TaskInfo`，并将其标记为空闲
      // 。
      worker[kTaskInfo].done(null, result);
      worker[kTaskInfo] = null;
      this.freeWorkers.push(worker);
      this.emit(kWorkerFreedEvent);
    });
    worker.on('error', (err) => {
      // 发生未捕获异常时：调用传递给
      // `runTask` 的回调并传入错误。
      if (worker[kTaskInfo])
        worker[kTaskInfo].done(err, null);
      else
        this.emit('error', err);
      // 从列表中移除 worker 并启动一个新的 Worker 来替换
      // 当前的 worker。
      this.workers.splice(this.workers.indexOf(worker), 1);
      this.addNewWorker();
    });
    this.workers.push(worker);
    this.freeWorkers.push(worker);
    this.emit(kWorkerFreedEvent);
  }

  runTask(task, callback) {
    if (this.freeWorkers.length === 0) {
      // 没有空闲线程，等待直到一个 worker 线程变为空闲。
      this.tasks.push({ task, callback });
      return;
    }

    const worker = this.freeWorkers.pop();
    worker[kTaskInfo] = new WorkerPoolTaskInfo(callback);
    worker.postMessage(task);
  }

  close() {
    for (const worker of this.workers) worker.terminate();
  }
}

import WorkerPool from './worker_pool.js';
import os from 'node:os';

const pool = new WorkerPool(os.availableParallelism());

let finished = 0;
for (let i = 0; i < 10; i++) {
  pool.runTask({ a: 42, b: 100 }, (err, result) => {
    console.log(i, err, result);
    if (++finished === 10)
      pool.close();
  });
}

import { createServer } from 'node:http';
import { AsyncResource, executionAsyncId } from 'node:async_hooks';

const server = createServer((req, res) => {
  req.on('close', AsyncResource.bind(() => {
    // 执行上下文绑定到当前的外部作用域。
  }));
  req.on('close', () => {
    // 执行上下文绑定到导致触发 'close' 的作用域。
  });
  res.end();
}).listen(3000);