Event Bus und das Observer-Pattern in Go (Golang): pub/sub-Implementierung

Event Bus und das Observer-Pattern in Go (Golang): von der Theorie zur Praxis

Das Observer-Pattern (Beobachter) und seine Implementierung als Event Bus (Ereignisbus) sind grundlegende Werkzeuge für den Aufbau lose gekoppelter, skalierbarer Systeme. In Go, mit seiner Philosophie „Kommuniziere über Kanäle, nicht über gemeinsamen Speicher“, gewinnen diese Patterns eine besondere Eleganz. In diesem Artikel werden wir untersuchen, wie man einen eigenen Event Bus in Go erstellt, unter Verwendung von Goroutinen und Kanälen, und typische Anwendungsszenarien betrachten.

Das Observer-Pattern: Kern und Problem

Stellen Sie sich ein System vor, in dem ein Objekt (Herausgeber) viele andere (Abonnenten) über eine Zustandsänderung benachrichtigen muss. Ohne Observer müssten Sie den Herausgeber fest mit jedem Abonnenten koppeln. Observer löst dies, indem es ein Abonnement-Interface einführt: Der Herausgeber speichert eine Liste von Abonnenten und ruft deren Update()-Methode auf, wenn ein Ereignis eintritt.

Event Bus ist ein zentraler Bus, der die Rolle des Vermittlers übernimmt. Jede Komponente kann ein Ereignis veröffentlichen, und jede andere kann es abonnieren, ohne voneinander zu wissen.

Warum ist Go ideal für einen Event Bus?

\n
• Goroutinen und Kanäle: Asynchrone Ereignisverarbeitung ohne Blockaden.
\n
• Interfaces: Ermöglichen die einfache Definition von Verträgen für Handler.
\n
• Keine Vererbung: Komposition über Kanäle und Funktionen ist natürlicher als klassische OOP-Implementierung.
\n

Einfache Implementierung eines Event Bus in Go

Beginnen wir mit einer Basisversion, bei der Ereignisse Strings sind und Abonnenten Handler-Funktionen.

package main\n\nimport (\n    "fmt"\n    "sync"\n    "time"\n)\n\n// EventBus — unser Ereignisbus\ntype EventBus struct {\n    subscribers map[string][]func(string)\n    mu          sync.RWMutex\n}\n\n// NewEventBus erstellt einen neuen Bus\nfunc NewEventBus() *EventBus {\n    return &EventBus{\n        subscribers: make(map[string][]func(string)),\n    }\n}\n\n// Subscribe abonniert einen Handler für ein Ereignis\nfunc (eb *EventBus) Subscribe(event string, handler func(string)) {\n    eb.mu.Lock()\n    defer eb.mu.Unlock()\n    eb.subscribers[event] = append(eb.subscribers[event], handler)\n}\n\n// Publish veröffentlicht ein Ereignis und ruft alle Abonnenten synchron auf\nfunc (eb *EventBus) Publish(event string, data string) {\n    eb.mu.RLock()\n    defer eb.mu.RUnlock()\n    if handlers, ok := eb.subscribers[event]; ok {\n        for _, handler := range handlers {\n            handler(data)\n        }\n    }\n}\n\nfunc main() {\n    bus := NewEventBus()\n    \n    bus.Subscribe("user.login", func(data string) {\n        fmt.Printf("Login: %s\\n", data)\n    })\n    \n    bus.Subscribe("user.login", func(data string) {\n        fmt.Printf("Audit: %s hat sich angemeldet\\n", data)\n    })\n    \n    bus.Publish("user.login", "alice@example.com")\n    // Ausgabe:\n    // Login: alice@example.com\n    // Audit: alice@example.com hat sich angemeldet\n}

Dieser Code funktioniert, aber es gibt einen Haken: Die Handler werden synchron aufgerufen. Wenn ein Handler hängt, blockiert er den gesamten Bus.

Asynchrone Version mit Kanälen

Fügen wir Asynchronität durch gepufferte Kanäle hinzu. Jeder Abonnent erhält seinen eigenen Kanal.

type AsyncEventBus struct {\n    subscribers map[string][]chan string\n    mu          sync.RWMutex\n}\n\nfunc NewAsyncEventBus() *AsyncEventBus {\n    return &AsyncEventBus{\n        subscribers: make(map[string][]chan string),\n    }\n}\n\nfunc (eb *AsyncEventBus) Subscribe(event string, buffer int) <-chan string {\n    eb.mu.Lock()\n    defer eb.mu.Unlock()\n    ch := make(chan string, buffer)\n    eb.subscribers[event] = append(eb.subscribers[event], ch)\n    return ch\n}\n\nfunc (eb *AsyncEventBus) Publish(event string, data string) {\n    eb.mu.RLock()\n    defer eb.mu.RUnlock()\n    if channe