Введение

Эти классы используются для связывания состояния и его передачи через обратные вызовы и цепочки промисов. Они позволяют хранить данные на протяжении всего времени обработки веб-запроса или любого другого асинхронного процесса. Это похоже на локальное хранилище потока в других языках.

Классы AsyncLocalStorage и AsyncResource входят в состав модуля node:async_hooks:

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

const { AsyncLocalStorage, AsyncResource } = require('node:async_hooks');

Класс: AsyncLocalStorage

Этот класс создаёт хранилища, которые сохраняют согласованность при асинхронных операциях.

Хотя можно создать собственную реализацию на основе модуля node:async_hooks, предпочтительнее использовать AsyncLocalStorage — это производительная реализация с безопасным управлением памятью, включающая существенные оптимизации, реализация которых неочевидна.

В следующем примере AsyncLocalStorage используется для создания простого регистратора, который присваивает идентификаторы входящим HTTP-запросам и добавляет их в сообщения, записываемые в журнал в рамках каждого запроса.

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');
    // 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
//   0: finish
//   1: start
//   1: finish

const http = require('node:http');
const { AsyncLocalStorage } = require('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');
    // 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
//   0: finish
//   1: start
//   1: finish

Каждый экземпляр AsyncLocalStorage поддерживает независимый контекст хранилища. Несколько экземпляров могут безопасно существовать одновременно без риска, что они будут мешать друг другу или данным друг друга.

new AsyncLocalStorage()

Создаёт новый экземпляр AsyncLocalStorage. Хранилище доступно только внутри вызова run() или после вызова enterWith().

Статический метод: AsyncLocalStorage.bind(fn)

• fn <Function> Функция, которую нужно связать с текущим контекстом выполнения.
• Возвращает: <Function> Новая функция, которая вызывает fn в сохранённом контексте выполнения.

Связывает указанную функцию с текущим контекстом выполнения.

Статический метод: AsyncLocalStorage.snapshot()

• Возвращает: <Function> Новая функция с сигнатурой (fn: (...args) : R, ...args) : R.

Сохраняет текущий контекст выполнения и возвращает функцию, принимающую функцию в качестве аргумента. При каждом вызове возвращённой функции она вызывает переданную ей функцию в сохранённом контексте.

const asyncLocalStorage = new AsyncLocalStorage();
const runInAsyncScope = asyncLocalStorage.run(123, () => AsyncLocalStorage.snapshot());
const result = asyncLocalStorage.run(321, () => runInAsyncScope(() => asyncLocalStorage.getStore()));
console.log(result);  // returns 123 copy

AsyncLocalStorage.snapshot() может заменить использование AsyncResource для простого отслеживания асинхронного контекста, например:

class Foo {
  #runInAsyncScope = AsyncLocalStorage.snapshot();

  get() { return this.#runInAsyncScope(() => asyncLocalStorage.getStore()); }
}

const foo = asyncLocalStorage.run(123, () => new Foo());
console.log(asyncLocalStorage.run(321, () => foo.get())); // returns 123 copy

asyncLocalStorage.disable()

Отключает экземпляр AsyncLocalStorage. Все последующие вызовы asyncLocalStorage.getStore() будут возвращать undefined, пока снова не будет вызван asyncLocalStorage.run() или asyncLocalStorage.enterWith().

При вызове asyncLocalStorage.disable() будут завершены все текущие контексты, связанные с экземпляром.

Вызов asyncLocalStorage.disable() необходим для того, чтобы asyncLocalStorage можно было удалить сборщиком мусора. Это не относится к хранилищам, предоставляемым asyncLocalStorage, поскольку эти объекты удаляются сборщиком мусора вместе с соответствующими асинхронными ресурсами.

Используйте этот метод, когда asyncLocalStorage больше не используется в текущем процессе.

asyncLocalStorage.getStore()

• Возвращает: <any>

Возвращает текущее хранилище. Если вызов выполнен вне асинхронного контекста, инициализированного вызовом asyncLocalStorage.run() или asyncLocalStorage.enterWith(), возвращается undefined.

asyncLocalStorage.enterWith(store)

• store <any>

Переходит в контекст на оставшуюся часть текущего синхронного выполнения, а затем сохраняет хранилище для всех последующих асинхронных вызовов.

Пример:

const store = { id: 1 };
// Replaces previous store with the given store object
asyncLocalStorage.enterWith(store);
asyncLocalStorage.getStore(); // Returns the store object
someAsyncOperation(() => {
  asyncLocalStorage.getStore(); // Returns the same object
}); copy

Этот переход продолжится на протяжении всего синхронного выполнения. Это означает, что, например, если контекст был установлен в обработчике события, последующие обработчики событий также будут выполняться в этом контексте, если только они не привязаны явно к другому контексту с помощью AsyncResource. Поэтому следует предпочитать run() вместо enterWith(), если нет веских причин использовать последний метод.

const store = { id: 1 };

emitter.on('my-event', () => {
  asyncLocalStorage.enterWith(store);
});
emitter.on('my-event', () => {
  asyncLocalStorage.getStore(); // Returns the same object
});

asyncLocalStorage.getStore(); // Returns undefined
emitter.emit('my-event');
asyncLocalStorage.getStore(); // Returns the same object copy

asyncLocalStorage.run(store, callback[, ...args])

• store <any>
• callback <Function>
• ...args <any>

Синхронно выполняет функцию в контексте и возвращает её результат. Хранилище недоступно за пределами функции обратного вызова. Оно доступно для любых асинхронных операций, созданных внутри функции обратного вызова.

Необязательные args передаются функции обратного вызова.

Если функция обратного вызова выдаёт ошибку, вызов run() также выдаёт эту ошибку. Этот вызов не влияет на трассировку стека, а контекст завершается.

Пример:

const store = { id: 2 };
try {
  asyncLocalStorage.run(store, () => {
    asyncLocalStorage.getStore(); // Returns the store object
    setTimeout(() => {
      asyncLocalStorage.getStore(); // Returns the store object
    }, 200);
    throw new Error();
  });
} catch (e) {
  asyncLocalStorage.getStore(); // Returns undefined
  // The error will be caught here
} copy

asyncLocalStorage.exit(callback[, ...args])

• callback <Function>
• ...args <any>

Синхронно выполняет функцию вне контекста и возвращает её результат. Хранилище недоступно внутри функции обратного вызова и асинхронных операций, созданных внутри неё. Любой вызов getStore(), выполненный внутри функции обратного вызова, всегда будет возвращать undefined.

Необязательные args передаются функции обратного вызова.

Если функция обратного вызова выдаёт ошибку, вызов exit() также выдаёт эту ошибку. Этот вызов не влияет на трассировку стека, а контекст восстанавливается.

Пример:

// Within a call to run
try {
  asyncLocalStorage.getStore(); // Returns the store object or value
  asyncLocalStorage.exit(() => {
    asyncLocalStorage.getStore(); // Returns undefined
    throw new Error();
  });
} catch (e) {
  asyncLocalStorage.getStore(); // Returns the same object or value
  // The error will be caught here
} copy

Использование с async/await

Если внутри асинхронной функции в контексте должен выполняться только один вызов await, следует использовать следующий шаблон:

async function fn() {
  await asyncLocalStorage.run(new Map(), () => {
    asyncLocalStorage.getStore().set('key', value);
    return foo(); // The return value of foo will be awaited
  });
} copy

В этом примере хранилище доступно только внутри функции обратного вызова и функций, вызываемых foo. За пределами run вызов getStore вернёт undefined.

Устранение неполадок: потеря контекста

В большинстве случаев AsyncLocalStorage работает без проблем. В редких случаях текущее хранилище теряется при выполнении одной из асинхронных операций.

Если ваш код основан на обратных вызовах, достаточно преобразовать их в промисы с помощью util.promisify(), чтобы код начал работать с нативными промисами.

Если необходимо использовать API с обратными вызовами или код предполагает пользовательскую реализацию thenable, используйте класс AsyncResource, чтобы связать асинхронную операцию с правильным контекстом выполнения. Найдите вызов функции, ответственный за потерю контекста, записывая содержимое asyncLocalStorage.getStore() после вызовов, которые, по вашему подозрению, приводят к потере. Если код выводит undefined, вероятно, за потерю контекста отвечает последний вызванный обратный вызов.

Класс: AsyncResource

Класс AsyncResource предназначен для расширения асинхронными ресурсами встраиваемой среды. С его помощью пользователи могут легко инициировать события жизненного цикла собственных ресурсов.

Хук init срабатывает при создании экземпляра AsyncResource.

Ниже приведён обзор API AsyncResource.

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

// AsyncResource() is meant to be extended. Instantiating a
// new AsyncResource() also triggers init. If triggerAsyncId is omitted then
// async_hook.executionAsyncId() is used.
const asyncResource = new AsyncResource(
  type, { triggerAsyncId: executionAsyncId(), requireManualDestroy: false },
);

// Run a function in the execution context of the resource. This will
// * establish the context of the resource
// * trigger the AsyncHooks before callbacks
// * call the provided function `fn` with the supplied arguments
// * trigger the AsyncHooks after callbacks
// * restore the original execution context
asyncResource.runInAsyncScope(fn, thisArg, ...args);

// Call AsyncHooks destroy callbacks.
asyncResource.emitDestroy();

// Return the unique ID assigned to the AsyncResource instance.
asyncResource.asyncId();

// Return the trigger ID for the AsyncResource instance.
asyncResource.triggerAsyncId();

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

// AsyncResource() is meant to be extended. Instantiating a
// new AsyncResource() also triggers init. If triggerAsyncId is omitted then
// async_hook.executionAsyncId() is used.
const asyncResource = new AsyncResource(
  type, { triggerAsyncId: executionAsyncId(), requireManualDestroy: false },
);

// Run a function in the execution context of the resource. This will
// * establish the context of the resource
// * trigger the AsyncHooks before callbacks
// * call the provided function `fn` with the supplied arguments
// * trigger the AsyncHooks after callbacks
// * restore the original execution context
asyncResource.runInAsyncScope(fn, thisArg, ...args);

// Call AsyncHooks destroy callbacks.
asyncResource.emitDestroy();

// Return the unique ID assigned to the AsyncResource instance.
asyncResource.asyncId();

// Return the trigger ID for the AsyncResource instance.
asyncResource.triggerAsyncId();

new AsyncResource(type[, options])

• type <string> Тип асинхронного события.
• options <Object>
  • triggerAsyncId <number> Идентификатор контекста выполнения, в котором было создано это асинхронное событие. По умолчанию: executionAsyncId().
  • requireManualDestroy <boolean> Если задано значение true, отключает emitDestroy при сборке объекта сборщиком мусора. Обычно это не требуется (даже если emitDestroy вызывается вручную), за исключением случаев, когда получено значение asyncId ресурса и с ним вызван emitDestroy чувствительного API. Если задано значение false, вызов emitDestroy при сборке мусора произойдёт только при наличии хотя бы одного активного хука destroy. По умолчанию: false.

Пример использования:

class DBQuery extends AsyncResource {
  constructor(db) {
    super('DBQuery');
    this.db = db;
  }

  getInfo(query, callback) {
    this.db.get(query, (err, data) => {
      this.runInAsyncScope(callback, null, err, data);
    });
  }

  close() {
    this.db = null;
    this.emitDestroy();
  }
} copy

Статический метод: AsyncResource.bind(fn[, type[, thisArg]])

• fn <Function> Функция, которую нужно связать с текущим контекстом выполнения.
• type <string> Необязательное имя для связывания с базовым AsyncResource.
• thisArg <any>

Связывает указанную функцию с текущим контекстом выполнения.

asyncResource.bind(fn[, thisArg])

• fn <Function> Функция, которую нужно связать с текущим AsyncResource.
• thisArg <any>

Связывает указанную функцию с областью видимости этого AsyncResource.

asyncResource.runInAsyncScope(fn[, thisArg, ...args])

• fn <Function> Функция, которую нужно вызвать в контексте выполнения этого асинхронного ресурса.
• thisArg <any> Получатель, который будет использоваться при вызове функции.
• ...args <any> Необязательные аргументы, передаваемые функции.

Вызывает указанную функцию с указанными аргументами в контексте выполнения асинхронного ресурса. При этом устанавливается контекст, вызываются хуки AsyncHooks перед обратными вызовами, выполняется функция, вызываются хуки AsyncHooks после обратных вызовов, а затем восстанавливается исходный контекст выполнения.

asyncResource.emitDestroy()

• Возвращает: <AsyncResource> Ссылка на asyncResource.

Вызывает все хуки destroy. Этот метод следует вызывать только один раз. При повторном вызове будет выдана ошибка. Этот метод необходимо вызвать вручную. Если ресурс будет удалён сборщиком мусора, хуки destroy вызваны не будут.

asyncResource.asyncId()

• Возвращает: <number> Уникальный asyncId, присвоенный ресурсу.

asyncResource.triggerAsyncId()

• Возвращает: <number> Тот же triggerAsyncId, который передаётся конструктору AsyncResource.

Использование AsyncResource для пула потоков Worker

В следующем примере показано, как использовать класс AsyncResource для корректного отслеживания асинхронных операций в пуле Worker. Аналогичным образом можно организовать другие пулы ресурсов, например пулы подключений к базе данных.

Предположим, что задача состоит в сложении двух чисел. Для этого используется файл task_processor.js со следующим содержимым:

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

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

Пул Worker вокруг него может иметь следующую структуру:

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`s are used only once.
  }
}

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();

    // Any time the kWorkerFreedEvent is emitted, dispatch
    // the next task pending in the queue, if any.
    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) => {
      // In case of success: Call the callback that was passed to `runTask`,
      // remove the `TaskInfo` associated with the Worker, and mark it as free
      // again.
      worker[kTaskInfo].done(null, result);
      worker[kTaskInfo] = null;
      this.freeWorkers.push(worker);
      this.emit(kWorkerFreedEvent);
    });
    worker.on('error', (err) => {
      // In case of an uncaught exception: Call the callback that was passed to
      // `runTask` with the error.
      if (worker[kTaskInfo])
        worker[kTaskInfo].done(err, null);
      else
        this.emit('error', err);
      // Remove the worker from the list and start a new Worker to replace the
      // current one.
      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) {
      // No free threads, wait until a worker thread becomes free.
      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();
  }
}

const { AsyncResource } = require('node:async_hooks');
const { EventEmitter } = require('node:events');
const path = require('node:path');
const { Worker } = require('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`s are used only once.
  }
}

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();

    // Any time the kWorkerFreedEvent is emitted, dispatch
    // the next task pending in the queue, if any.
    this.on(kWorkerFreedEvent, () => {
      if (this.tasks.length > 0) {
        const { task, callback } = this.tasks.shift();
        this.runTask(task, callback);
      }
    });
  }

  addNewWorker() {
    const worker = new Worker(path.resolve(__dirname, 'task_processor.js'));
    worker.on('message', (result) => {
      // In case of success: Call the callback that was passed to `runTask`,
      // remove the `TaskInfo` associated with the Worker, and mark it as free
      // again.
      worker[kTaskInfo].done(null, result);
      worker[kTaskInfo] = null;
      this.freeWorkers.push(worker);
      this.emit(kWorkerFreedEvent);
    });
    worker.on('error', (err) => {
      // In case of an uncaught exception: Call the callback that was passed to
      // `runTask` with the error.
      if (worker[kTaskInfo])
        worker[kTaskInfo].done(err, null);
      else
        this.emit('error', err);
      // Remove the worker from the list and start a new Worker to replace the
      // current one.
      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) {
      // No free threads, wait until a worker thread becomes free.
      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();
  }
}

module.exports = WorkerPool;

Без явного отслеживания, добавленного объектами WorkerPoolTaskInfo, могло бы показаться, что обратные вызовы связаны с отдельными объектами Worker. Однако создание объектов Workers не связано с созданием задач и не позволяет определить, когда задачи были запланированы.

Этот пул можно использовать следующим образом:

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();
  });
}

const WorkerPool = require('./worker_pool.js');
const os = require('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();
  });
}

Интеграция AsyncResource с EventEmitter

Обработчики событий, вызванные объектом EventEmitter, могут выполняться в контексте выполнения, отличном от того, который был активен при вызове eventEmitter.on().

В следующем примере показано, как использовать класс AsyncResource, чтобы связать обработчик события с правильным контекстом выполнения. Тот же подход можно применить к Stream или аналогичному классу, управляемому событиями.

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

const server = createServer((req, res) => {
  req.on('close', AsyncResource.bind(() => {
    // Execution context is bound to the current outer scope.
  }));
  req.on('close', () => {
    // Execution context is bound to the scope that caused 'close' to emit.
  });
  res.end();
}).listen(3000);

const { createServer } = require('node:http');
const { AsyncResource, executionAsyncId } = require('node:async_hooks');

const server = createServer((req, res) => {
  req.on('close', AsyncResource.bind(() => {
    // Execution context is bound to the current outer scope.
  }));
  req.on('close', () => {
    // Execution context is bound to the scope that caused 'close' to emit.
  });
  res.end();
}).listen(3000);