Ein einfaches Pub/Sub-System in Node.js bauen

Publish/Subscribe entkoppelt den Code, der ein Ereignis erzeugt, von dem Code, der darauf reagiert – der Publisher weiß nicht und es interessiert ihn nicht, wer zuhört, und die Subscriber wissen nicht und es interessiert sie nicht, woher das Ereignis kam. Zu einem Message Broker wie Redis oder RabbitMQ greift man, wenn man das prozessübergreifend braucht, aber das Muster selbst ist klein genug, um es im Prozess von Grund auf zu bauen, und das Verständnis dieser Version macht die Broker-gestützten viel weniger mysteriös.

Die Kernidee

Ein Pub/Sub-System hat drei bewegliche Teile: Topics (benannte Kanäle), Subscriber (Funktionen, die für ein Topic registriert sind) und eine Publish-Operation, die jeden aktuell für ein Topic registrierten Subscriber aufruft.

class EventBus {
  #topics = new Map();

  subscribe(topic, handler) {
    if (!this.#topics.has(topic)) {
      this.#topics.set(topic, new Set());
    }
    this.#topics.get(topic).add(handler);

    // return an unsubscribe function — the caller owns cleanup
    return () => this.#topics.get(topic)?.delete(handler);
  }

  publish(topic, payload) {
    const handlers = this.#topics.get(topic);
    if (!handlers) return;
    for (const handler of handlers) {
      handler(payload);
    }
  }
}

Das ist ein funktionierender In-Memory-Event-Bus in weniger als 20 Zeilen. Node's eingebauter EventEmitter leistet dasselbe, aber sich einmal einen eigenen zu bauen lohnt sich für die Dinge, die ein vollständiger Broker obendrauf legt: asynchrone Verarbeitung, Fehlerisolation und Wildcard-Topics.

Verwendung

const bus = new EventBus();

const unsubscribe = bus.subscribe('order.created', (order) => {
  console.log(`Sending confirmation email for order ${order.id}`);
});

bus.subscribe('order.created', (order) => {
  console.log(`Reserving inventory for order ${order.id}`);
});

bus.publish('order.created', { id: 42, total: 89.99 });
// Sending confirmation email for order 42
// Reserving inventory for order 42

unsubscribe(); // the email handler stops receiving events

Zwei unabhängige Belange – E-Mail und Lagerbestand – reagieren auf dasselbe Ereignis, ohne voneinander zu wissen. Fügt man später einen dritten Belang hinzu (etwa Analytics), abonniert man ihn, ohne den Code anzufassen, der order.created veröffentlicht.

Asynchrone Subscriber behandeln

Das naive publish oben ruft Handler synchron auf und ignoriert, ob sie ein Promise zurückgeben. Macht ein Handler asynchrone Arbeit (in eine Datenbank schreiben, eine API aufrufen), will man meist wissen, ob er fehlgeschlagen ist, und man will nicht, dass ein langsamer Handler die anderen blockiert.

async publish(topic, payload) {
  const handlers = this.#topics.get(topic);
  if (!handlers) return [];

  const results = await Promise.allSettled(
    [...handlers].map((handler) => handler(payload)),
  );

  for (const result of results) {
    if (result.status === 'rejected') {
      console.error('Subscriber failed:', result.reason);
    }
  }

  return results;
}

Promise.allSettled führt jeden Handler nebenläufig aus und lässt jeden unabhängig erfolgreich sein oder scheitern – ein kaputter Subscriber hindert die anderen nicht am Laufen, und man sieht, welcher gescheitert ist, statt dass eine unbehandelte Rejection den Fehler verschluckt.

Wildcard- und Namespace-Topics

Echte Systeme wollen oft eine ganze Familie von Ereignissen abonnieren (order.*) statt eines exakten Topics. Ein einfacher Ansatz: Subscriber unter exakten Topic-Strings speichern, beim Publish aber sowohl exakte als auch Wildcard-Treffer prüfen.

class EventBus {
  #topics = new Map();

  subscribe(pattern, handler) {
    if (!this.#topics.has(pattern)) this.#topics.set(pattern, new Set());
    this.#topics.get(pattern).add(handler);
    return () => this.#topics.get(pattern)?.delete(handler);
  }

  #matches(pattern, topic) {
    if (pattern === topic) return true;
    if (!pattern.endsWith('.*')) return false;
    const prefix = pattern.slice(0, -1); // keep the trailing dot
    return topic.startsWith(prefix);
  }

  async publish(topic, payload) {
    const matched = [...this.#topics.entries()]
      .filter(([pattern]) => this.#matches(pattern, topic))
      .flatMap(([, handlers]) => [...handlers]);

    return Promise.allSettled(matched.map((handler) => handler(payload)));
  }
}

Jetzt fängt bus.subscribe('order.*', logAnyOrderEvent) order.created, order.cancelled und alles andere unter dem Namespace order. ab, neben Handlern, die das exakte Topic abonniert haben.

Wo das nicht mehr reicht – und man einen echten Broker braucht

Dieser Bus im Prozess ist wirklich nützlich, um Module innerhalb einer einzelnen Anwendung zu entkoppeln. Er stößt an Grenzen, sobald man eines der folgenden Dinge braucht, und dort übernehmen Redis Pub/Sub, RabbitMQ oder Kafka:

• Prozess- oder dienstübergreifende Zustellung. Dein Event-Bus existiert nur innerhalb eines Node-Prozesses. Zwei Instanzen deiner App hinter einem Load Balancer teilen keine Subscriber.
• Dauerhaftigkeit. Ist beim Publish niemand abonniert, ist das obige Ereignis einfach weg. Ein echter Broker (oder zumindest eine Queue) persistiert Nachrichten, sodass ein später startender Subscriber sie trotzdem noch erhalten kann.
• Backpressure und Reihenfolgegarantien. Promise.allSettled gibt dir Nebenläufigkeit, keine Flusskontrolle – ein langsamer Konsument wird nicht gedrosselt, er fällt einfach zurück, während Produzenten weiter feuern.
• At-least-once-Zustellung mit Bestätigung. Echte Message Broker lassen einen Konsumenten eine Nachricht explizit quittieren (ack) und stellen sie erneut zu, wenn der Konsument vor dem Quittieren abstürzt.

Wann die Variante im Prozess die richtige Wahl ist

Befinden sich alle deine „Subscriber" in einem Node-Prozess – einen Webhook-Handler von einem E-Mail-Versender von einem Audit-Logger zu entkoppeln, alles in einem Dienst –, ist ein kleiner EventBus wie dieser oft genau die richtige Menge Infrastruktur. Kafka zu nehmen, um zwei Funktionen in derselben Codebasis zu verbinden, löst ein Problem, das man noch nicht hat. Fang hier an und steige auf einen Broker um, wenn du die Zustellung über Prozessgrenzen hinweg wirklich brauchst, nicht vorher.