Event Bus и паттерн Observer в Go (Golang)

Event Bus и паттерн Observer в Go (Golang): от теории к практике

Паттерн Observer (наблюдатель) и его реализация в виде Event Bus (шины событий) — фундаментальные инструменты для построения слабосвязанных, масштабируемых систем. В Go, с его философией «общайтесь через каналы, а не через разделяемую память», эти паттерны приобретают особую элегантность. В этой статье мы разберем, как создать собственный Event Bus на Go, используя горутины и каналы, и рассмотрим типовые сценарии использования.

Паттерн Observer: суть и проблема

Представьте, что у вас есть система, где один объект (издатель) должен уведомлять множество других (подписчиков) об изменении своего состояния. Без Observer вам пришлось бы жестко связывать издателя с каждым подписчиком. Observer решает это, вводя интерфейс подписки: издатель хранит список подписчиков и вызывает у них метод Update() при наступлении события.

Event Bus — это централизованная шина, которая берет на себя роль посредника. Любой компонент может опубликовать событие, а любой другой — подписаться на него, не зная о существовании друг друга.

Почему Go идеален для Event Bus?

Горутины и каналы: асинхронная обработка событий без блокировок.
Интерфейсы: позволяют легко определять контракты для обработчиков.
Отсутствие наследования: композиция через каналы и функции естественнее, чем классическая ООП-реализация.

Простая реализация Event Bus на Go

Начнем с базовой версии, где события — это строки, а подписчики — функции-обработчики.

package main

import (
    "fmt"
    "sync"
    "time"
)

// EventBus — наша шина событий
type EventBus struct {
    subscribers map[string][]func(string)
    mu          sync.RWMutex
}

// NewEventBus создает новую шину
func NewEventBus() *EventBus {
    return &EventBus{
        subscribers: make(map[string][]func(string)),
    }
}

// Subscribe подписывает обработчик на событие
func (eb *EventBus) Subscribe(event string, handler func(string)) {
    eb.mu.Lock()
    defer eb.mu.Unlock()
    eb.subscribers[event] = append(eb.subscribers[event], handler)
}

// Publish публикует событие и вызывает всех подписчиков синхронно
func (eb *EventBus) Publish(event string, data string) {
    eb.mu.RLock()
    defer eb.mu.RUnlock()
    if handlers, ok := eb.subscribers[event]; ok {
        for _, handler := range handlers {
            handler(data)
        }
    }
}

func main() {
    bus := NewEventBus()
    
    bus.Subscribe("user.login", func(data string) {
        fmt.Printf("Логин: %s\n", data)
    })
    
    bus.Subscribe("user.login", func(data string) {
        fmt.Printf("Аудит: %s вошел в систему\n", data)
    })
    
    bus.Publish("user.login", "alice@example.com")
    // Вывод:
    // Логин: alice@example.com
    // Аудит: alice@example.com вошел в систему
}

Этот код работает, но есть нюанс: обработчики вызываются синхронно. Если один обработчик зависнет, он заблокирует всю шину.

Асинхронная версия с каналами

Добавим асинхронность через буферизированные каналы. Каждый подписчик будет получать свой канал.

type AsyncEventBus struct {
    subscribers map[string][]chan string
    mu          sync.RWMutex
}

func NewAsyncEventBus() *AsyncEventBus {
    return &AsyncEventBus{
        subscribers: make(map[string][]chan string),
    }
}

func (eb *AsyncEventBus) Subscribe(event string, buffer int) <-chan string {
    eb.mu.Lock()
    defer eb.mu.Unlock()
    ch := make(chan string, buffer)
    eb.subscribers[event] = append(eb.subscribers[event], ch)
    return ch
}

func (eb *AsyncEventBus) Publish(event string, data string) {
    eb.mu.RLock()
    defer eb.mu.RUnlock()
    if channels, ok := eb.subscribers[event]; ok {
        for _, ch := range channels {
            select {
            case ch <- data:
            default:
                // Если канал полон, пропускаем (non-blocking)
            }
        }
    }
}

// Использование
func main() {
    bus := NewAsyncEventBus()
    ch := bus.Subscribe("order.created", 10)
    
    go func() {
        for data := range ch {
            fmt.Println("Получен заказ:", data)
            time.Sleep(100 * time.Millisecond) // имитация работы
        }
    }()
    
    bus.Publish("order.created", "Заказ #123")
    time.Sleep(200 * time.Millisecond)
}