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

Publish/Subscribe entkoppelt den Code, der a Event erzeugt, von dem Code, der drauf reagiert - da Publisher woaß ned und is a wurscht, wer zuhört, und de Subscriber wissn ned und ist's wurscht, woher des Event kummt. Du greifst zu am Message Broker wia Redis oda RabbitMQ, sobois des üba Prozessgrenzen hinweg brauchst, aber des Muster söba is klan gnua, dass ma's in-process von Grund auf bauen kann, und wer de Version vasteht, find't die Broker-basierten viel weniger rätselhaft.

Da Kerngedanke

A Pub/Sub-System hot drei bewegliche Teile: Topics (benannte Kanäle), Subscriber (Funktionen, de auf an Topic registriert sand) und a Publish-Operation, de jeden grod registrierten Subscriber für an Topic aufruaft.

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

Des is a funktionierender In-Memory-Event-Bus in unter 20 Zeiln. Nodes eingebauter EventEmitter mocht den söbn Job, aber söba amoi einen zum bauen zahlt si aus, wegen de Teile, de a vollwertiger Broker obendrauf bringt: asynchrone Behandlung, Fehlerisolierung und Wildcard-Topics.

Valwendung

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

Zwoa unabhängige Belange - E-Mail und Lagerbestand - reagiern auf des söbe Event, ohne voneinander zum wissn. Fügst später an dritten Belang dazu (sagn ma Analytics), subscribst ihn, ohne den Code zum anfassn, der order.created publisht.

Asynchrone Subscriber behandeln

Des naive publish von oben ruaft Handler synchron auf und ignoriert, ob's a Promise zruckgebn. Wenn a Handler asynchron orbeitet (in a Datenbank schreibn, a API aufrufn), willst normalerweise wissn, ob's gscheitert is, und du wüst ned, dass oa langsamer Handler de ondern 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 lasst olle Handler gleichzeitig laufn und jeden unabhängig erfolgreich sei oda scheitern - a kaputter Subscriber hindert de ondern ned am Laufn, und du siegst, wöcher gscheitert is, statt dass a unhandled Rejection den Fehler verschluckt.

Wildcard- und Namespace-Topics

Echte Systeme wolln oft a ganze Familie vo Events abonniern (order.*) statt oan exakten Topic. Einfacher Ansatz: Subscriber unter exakten Topic-Strings speichern, aber beim Publish sowohl exakte ois a Wildcard-Treffer prüfn.

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 olles andere unterm order.-Namespace, neben den Handlern, de auf den exakten Topic subscribt hom.

Wo des ned mehr reicht - und du an echtn Broker brauchst

Dieser In-Process-Bus is wirklich nützlich, um Module innerhoib von ana einzelnen Anwendung zum entkoppeln. Er stößt an seine Grenzen, sobois eins vo dem brauchst, und do übernehmen Redis Pub/Sub, RabbitMQ oda Kafka:

• Zustellung üba Prozess- oda Servicegrenzen. Dein Event-Bus gibt's nur in oam Node-Prozess. Zwoa Instanzen deiner App hinter am Load Balancer teiln si koane Subscriber.
• Dauerhaftigkeit. Wenn beim Publish niemand subscribt hot, is des Event oben einfach weg. A echter Broker (oder zumindest a Queue) speichert Nachrichten, damit a später gstarteter Subscriber sie noch kriagt.
• Backpressure und Reihenfolge-Garantien. Promise.allSettled gibt dir Nebenläufigkeit, koa Flusssteuerung - a langsamer Consumer wird ned gedrosselt, er bleibt einfach zruck, während de Producer weiterfeiern.
• At-least-once-Zustellung mit Bestätigung. Echte Message Broker lossn an Consumer a Nachricht explizit bestätigen und stöin sie nochmal zu, wenn da Consumer vorm Bestätigen obstürzt.

Wann de In-Process-Version richtig is

Wenn deine "Subscriber" olle innerhoib vo oam Node-Prozess liegn - an Webhook-Handler von an E-Mail-Sender von an Audit-Logger entkoppeln, olles in oam Service -, dann is a klaner EventBus wia dieser oft genau de richtige Menge an Infrastruktur. Kafka rausholn, um zwoa Funktionen in derselben Codebasis zu verbinden, heißt, a Problem zum lösn, des du no gar ned host. Fang do an und wechsl zu am Broker, wennst wirklich Zustellung üba Prozessgrenzen brauchst - ned früher, du Hirnederl.