EventEmitter в Node.js: события своими руками

Половина API самой Node построена на событиях: HTTP-сервер эмитит request, поток — data и end, дочерний процесс — close, сам процесс — exit. Всё это один и тот же класс EventEmitter из модуля node:events, и его можно использовать в своём коде.

Что такое EventEmitter простыми словами

EventEmitter — это объект-посредник со списком подписчиков. Одни функции регистрируют интерес к событию по имени, другие сообщают, что событие произошло. Ни те, ни другие не знают друг о друге — связывает их только строка с именем события.

Именно поэтому события применяют там, где число реакций заранее неизвестно: на одно «пользователь зарегистрировался» может быть подписана и отправка письма, и запись в аналитику, и выдача бонуса — и добавление четвёртой реакции не трогает код регистрации.

Основы: on и emit

import { EventEmitter } from 'node:events';

const bus = new EventEmitter();

bus.on('order', (id) => console.log('слушатель 1:', id));
bus.on('order', (id) => console.log('слушатель 2:', id));

console.log('emit вернул:', bus.emit('order', 'A-1'));
console.log('слушателей :', bus.listenerCount('order'));
console.log('emit без слушателей:', bus.emit('nobody'));

слушатель 1: A-1
слушатель 2: A-1
emit вернул: true
слушателей : 2
emit без слушателей: false

Что важно в этом выводе:

• Обработчики вызываются в порядке подписки — сначала первый, потом второй.
• emit возвращает true, если хоть кто-то слушал, и false, если подписчиков не было. Событие при этом просто исчезает: никакой очереди и никакой ошибки.
• Аргументы после имени события передаются обработчикам как есть.

emit синхронен — и это главный подвох

Вопреки распространённому мнению, emit не откладывает вызов. Обработчики выполняются немедленно, в том же стеке, до следующей строки после emit:

const sync = new EventEmitter();
sync.on('ping', () => console.log('  2. обработчик'));

console.log('  1. до emit');
sync.emit('ping');
console.log('  3. после emit');

  1. до emit
  2. обработчик
  3. после emit

Два следствия, о которых стоит помнить:

1. Медленный обработчик тормозит того, кто вызвал emit. Если внутри обработчика тяжёлые вычисления, они блокируют событийный цикл — как и любой синхронный код.
2. Асинхронный обработчик никто не ждёт. Если передать в on функцию с async, emit вернёт управление, не дожидаясь её завершения, а ошибка внутри станет unhandledRejection. Ловите ошибки внутри самого обработчика.

once: подписка на один раз

once снимает обработчик сразу после первого срабатывания — удобно для событий, которые случаются единожды: «соединение установлено», «конфиг загружен».

const o = new EventEmitter();
o.once('ready', () => console.log('сработал один раз'));

o.emit('ready');
o.emit('ready');
console.log('слушателей после:', o.listenerCount('ready'));

сработал один раз
слушателей после: 0

Второй emit не напечатал ничего, а счётчик слушателей обнулился сам.

events.once: событие как промис

Это функция уровня модуля, не метод объекта. Она возвращает промис, который разрешается при следующем срабатывании события — так событийный код встраивается в обычный async/await:

import { EventEmitter, once } from 'node:events';

const later = new EventEmitter();
setTimeout(() => later.emit('done', 'значение из события'), 20);

const [value] = await once(later, 'done');
console.log('await once ->', value);

await once -> значение из события

Обратите внимание на деструктуризацию [value]: once отдаёт массив аргументов события, потому что их может быть несколько.

Это рабочий способ подружить событийный API с промисами — например, дождаться завершения дочернего процесса обычным await, а не колбэком.

events.on: события как поток в for await

У once есть старший брат — функция on. Она превращает эмиттер в асинхронный итератор: события можно перебирать циклом, как элементы массива.

import { EventEmitter, on } from 'node:events';

const bus = new EventEmitter();

(async () => {
  for (let i = 1; i <= 3; i++) {
    await sleep(10);
    bus.emit('job', `задача-${i}`);
  }
  bus.emit('job', null);   // сигнал остановки
})();

const seen = [];
for await (const [payload] of on(bus, 'job')) {
  if (payload === null) break;
  seen.push(payload);
}
console.log('обработано по очереди:', seen.join(', '));

обработано по очереди: задача-1, задача-2, задача-3

Зачем это нужно, если есть on(event, handler)? Разница в управлении потоком. Обработчик через .on() вызывается на каждое событие немедленно, и если внутри асинхронная работа, обработчики наслаиваются друг на друга — десять событий подряд запустят десять параллельных обработок. В for await следующая итерация не начнётся, пока не закончилась предыдущая.

Практический выбор:

Важная деталь: цикл for await ... of on() не завершается сам. Эмиттер не знает, что события кончились, поэтому выход нужно предусмотреть — по значению-сигналу, как выше, либо через AbortSignal в опциях on().

Событие error — особый случай

error — единственное событие со специальной семантикой. Если его эмитят, а обработчика нет, EventEmitter бросает исключение:

const risky = new EventEmitter();

try {
  risky.emit('error', new Error('никто не слушает error'));
} catch (err) {
  console.log('emit("error") бросил:', err.message);
}

emit("error") бросил: никто не слушает error

Без try/catch это необработанное исключение — то есть падение процесса. Отсюда правило: на любой эмиттер, который может сообщить об ошибке, вешайте on('error'). Это касается и встроенных объектов — сокетов, потоков, серверов.

Утечка слушателей

Node предупредит, если на одно событие подписано больше 10 обработчиков:

MaxListenersExceededWarning: Possible EventEmitter memory leak detected.
11 order listeners added. Use emitter.setMaxListeners() to increase limit

Это не ошибка, а диагностика. Почти всегда причина в том, что подписку делают в цикле или при каждом запросе, а снимать забывают. Снимается она через off — той же функцией, что подписывались:

const handler = (id) => console.log(id);
bus.on('order', handler);
bus.off('order', handler);   // анонимную функцию так снять невозможно

Поднимать лимит через setMaxListeners осмысленно, только если 11+ подписчиков — осознанное решение архитектуры, а не забытый off.

Свой класс на событиях

Наследование от EventEmitter даёт объекту on/emit бесплатно:

class Queue extends EventEmitter {
  #items = [];

  add(task) {
    this.#items.push(task);
    this.emit('added', task, this.#items.length);
  }
}

const q = new Queue();
q.on('added', (task, size) => console.log(`добавлено ${task}, в очереди ${size}`));
q.add('отчёт');

Смежные темы: промисы — чем событие отличается от однократного результата; потоки — самые используемые эмиттеры в платформе; модуль fs — где события всплывают при работе с файлами. Полный список — в справочнике по Node.js.