events: EventEmitter

EventEmitter — базовый класс Node.js для реализации паттерна Observer (Pub/Sub): позволяет объектам генерировать именованные события и регистрировать слушателей (handlers) на эти события.

Зачем нужно

EventEmitter — основа асинхронного программирования в Node.js. Все ключевые объекты наследуют от него: Stream, http.Server, process, fs.ReadStream. Понимание EventEmitter необходимо для работы с потоками, отладки memory leak (слишком много слушателей) и построения event-driven архитектуры внутри Node.js-сервиса.

Где используется

  • Все Stream-объекты ('data', 'end', 'error')
  • http.Server — события 'request', 'close'
  • process — 'exit', 'uncaughtException', 'SIGTERM'
  • Кастомные event bus внутри приложения
  • Worker Threads IPC

Основной контент

Базовый EventEmitter

const EventEmitter = require('events');

const emitter = new EventEmitter();

// Зарегистрировать обработчик
emitter.on('data', (payload) => {
  console.log('Получено:', payload);
});

// Одноразовый обработчик
emitter.once('connect', () => {
  console.log('Подключено (только один раз)');
});

// Сгенерировать событие
emitter.emit('data', { id: 1, message: 'Hello' });
emitter.emit('connect');
emitter.emit('connect'); // повторно — обработчик не вызовется

// Удалить обработчик
function handler(data) { console.log(data); }
emitter.on('event', handler);
emitter.off('event', handler);         // удалить
emitter.removeAllListeners('event');   // удалить все

// Количество обработчиков
console.log(emitter.listenerCount('data')); // 1

Наследование от EventEmitter

const EventEmitter = require('events');

class OrderService extends EventEmitter {
  constructor(db) {
    super;
    this.db = db;
  }

  async create(orderData) {
    const order = await this.db.orders.create(orderData);

    // Уведомить подписчиков
    this.emit('order:created', order);

    return order;
  }

  async cancel(orderId) {
    await this.db.orders.update({ id: orderId, status: 'cancelled' });
    this.emit('order:cancelled', { orderId });
  }
}

// Использование
const orderService = new OrderService(db);

// Подписаться на события
orderService.on('order:created', async (order) => {
  await emailService.sendConfirmation(order);
});

orderService.on('order:created', async (order) => {
  await inventoryService.reserve(order.items);
});

orderService.on('order:cancelled', async ({ orderId }) => {
  await inventoryService.release(orderId);
});

Event Bus — глобальная шина событий

// events/bus.js
const EventEmitter = require('events');

class EventBus extends EventEmitter {
  constructor {
    super;
    this.setMaxListeners(50); // поднять лимит
  }
}

module.exports = new EventBus(); // singleton

// Публикатор
const bus = require('./events/bus');
bus.emit('payment:completed', { orderId: 42, amount: 999 });

// Подписчик (в другом модуле)
const bus = require('./events/bus');
bus.on('payment:completed', async ({ orderId }) => {
  await OrderService.markAsPaid(orderId);
});

Async обработчики и ошибки

// EventEmitter не поддерживает async из коробки
// Ошибки в async-обработчиках не перехватываются

// НЕПРАВИЛЬНО:
emitter.on('data', async (data) => {
  throw new Error('Это не поймает emitter'); // unhandledRejection!
});

// ПРАВИЛЬНО — оборачивать в try/catch:
emitter.on('data', async (data) => {
  try {
    await processData(data);
  } catch (err) {
    emitter.emit('error', err); // перенаправить в error handler
  }
});

// Обязательный error handler:
emitter.on('error', (err) => {
  console.error('Emitter error:', err);
  // без этого Node.js упадёт с необработанным исключением!
});

Полезные методы

// Слушать один раз
emitter.once('ready', () => console.log('ready!'));

// Получить список событий
console.log(emitter.eventNames); // ['data', 'error']

// Получить обработчики
const handlers = emitter.listeners('data');

// Максимум слушателей (по умолчанию 10)
// При превышении: MaxListenersExceededWarning (не ошибка)
emitter.setMaxListeners(20);

Частые ошибки

  • Нет обработчика 'error' — если объект emits 'error' и нет listener, Node.js бросает исключение и падает; всегда добавлять emitter.on('error', handler)
  • Memory Leak — добавлять обработчики в цикле или при каждом запросе без removeListener; превышение maxListeners — сигнал утечки
  • Async-обработчики без try/catch — rejection не перехватывается EventEmitter
  • Забыть вызвать removeAllListeners — при уничтожении объекта оставшиеся listeners держат ссылку, препятствуя GC

Связанные темы

Ресурсы

🎓 Источник: Летняя школа 2017 — Файлы, потоки, буферы, сеть, сокеты, ошибки

  • 📅 2019-11-17 · YouTube · Cvsv672Lbvk
  • Тезисы:
    • EventEmitter — это ~10 строк кода. Хватает двух методов: on и emit. Понять = написать самому
    • on(name, fn) — если массив для этого события пустой, аллоцируется новый и в него push; emit(name, data) — forEach по массиву
    • События — это «GoTo для JS»: передача управления куда угодно. Развязка частей программы, можно несколько подписчиков или ни одного
    • Перехват всех событий (wildcard *): сохранить оригинальный emit в замыкание, в обёртке делать два apply — по имени и по *
  • Код (минимальный):
    class EventEmitter {
  constructor { this.events = {}; }
  on(name, fn) {
    const list = this.events[name] || (this.events[name] = );
    list.push(fn);
  }
  emit(name, data) {
    const list = this.events[name];
    if (list) list.forEach(fn => fn(data));
  }
}
  • Цитата:

    «События — это GoTo для JS. Передача управления куда угодно. Место для ошибок.»

🎓 Источник: Декомпозиция логера, доработка EventEmitter (Летняя школа 2022)

  • 📅 2022-08-25 · YouTube
  • Тезисы: расширение EventEmitter promise-обёртками, async-friendly API
  • 📅 2022-08-26 · YouTube
    • Ревью и рефакторинг metalog и metacom, EventEmitter в metarhia