Event-driven архитектура на Go: Строим реактивные системы будущего

Почему Event-driven архитектура на Go — это новый черный пояс в разработке?

Представьте себе оркестр, где каждый музыкант играет свою партию, не дожидаясь дирижера, но при этом создает идеальную симфонию. Именно так работает event-driven архитектура на Go — каждый сервис живет своей жизнью, реагирует на события и производит новые, не блокируя общий поток.

В мире, где пользователи требуют мгновенной реакции, а системы обрабатывают миллионы запросов в секунду, синхронные REST API начинают "захлебываться". Go с его легковесными горутинами и нативной поддержкой конкурентности становится идеальным языком для построения реактивных систем. Но как превратить хаос асинхронных вызовов в стройную архитектуру? Давайте разбираться.

Что такое Event-driven архитектура (EDA) и почему Go?

Event-driven архитектура на Go — это парадигма, где компоненты системы общаются через события, а не прямые вызовы. Вместо того чтобы сервис A ждал ответа от сервиса B, он просто публикует сообщение в брокер, а B подхватывает его, когда будет готов.

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

• Горутины — дешевые "потоки" для параллельной обработки событий (можно запустить миллион без проблем)
• Каналы — встроенный механизм для передачи событий между горутинами
• Быстрая компиляция — итерации в микросервисах становятся молниеносными
• Простота — меньше "магии", больше контроля

Строим первую event-driven систему на Go: от простого к сложному

Уровень 1: Внутренние события через каналы

Начнем с основ. Представьте, что мы пишем систему мониторинга сервера. Каждое событие (падение памяти, рост CPU) должно обрабатываться асинхронно.

package main

import (
    "fmt"
    "math/rand"
    "time"
)

type Event struct {
    Type    string
    Payload interface{}
}

func main() {
    // Создаем канал для событий
    eventChannel := make(chan Event, 100)
    
    // Продюсер: генерируем события
    go func() {
        for {
            time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
            event := Event{
                Type: "system.memory.alert",
                Payload: map[string]interface{}{
                    "usage_pct": rand.Float64() * 100,
                    "timestamp": time.Now(),
                },
            }
            eventChannel <- event
        }
    }()
    
    // Консьюмер: обрабатываем события
    go func() {
        for event := range eventChannel {
            fmt.Printf("Получено событие: %s\n", event.Type)
            processEvent(event)
        }
    }()
    
    time.Sleep(5 * time.Second)
}

func processEvent(e Event) {
    // Имитация обработки
    time.Sleep(200 * time.Millisecond)
    fmt.Printf("Обработано: %v\n", e.Payload)
}

Это база, но для продакшена нужно нечто большее. Когда количество сервисов переваливает за десяток, на сцену выходят брокеры сообщений.

Уровень 2: Event-driven архитектура на Go с NATS

NATS — легковесный брокер, который идеально дружит с Go. Он быстрее RabbitMQ и проще Kafka.

package main

import (
    "log"
    "time"
    "github.com/nats-io/nats.go"
)

type OrderEvent struct {
    OrderID string    `json:"order_id"`
    Status  string    `json:"status"`
    Amount  float64   `json:"amount"`
    Time    time.Time `json:"time"`
}

func main() {
    // Подключаемся к NATS
    nc, err := nats.Connect(nats.DefaultURL)
    if err != nil {
        log.Fatal(err)
    }
    defer nc.Close()
    
    // Публикуем событие (продюсер)
    event := OrderEvent{
        OrderID: "ORD-12345",
        Status:  "created",
        Amount:  299.99,
        Time:    time.Now(),
    }
    
    data, _ := json.Marshal(event)
    nc.Publish("orders.created", data)
    
    // Подписываемся на события (консьюмер)
    nc.Subscribe("orders.*", func(msg *nats.Msg) {
        var order OrderEvent
        json.Unmarshal(msg.Data, &order)
        log.Printf("Получен заказ: %s, статус: %s", order.OrderID, order.Status)
        
        // Асинхронная обработка
        go processOrder(order)
    })
    
    // Блокируем main
    select {}
}

func processOrder(order OrderEvent) {
    // Эмуляция сложной бизнес-логики
    time.Sleep(1 * time.Second)
    log.Printf("Заказ %s обработан", order.OrderID)
}

Уровень 3: Продвинутая обработка с Kafka на Go

Когда нужна гарантированная доставка и реплей событий, Kafka становится незаменимой. Разберем event-driven архитектуру на Go с Kafka.

package main

import (
    "context"
    "log"
    "time"
    "github.com/segmentio/kafka-go"
)

func main() {
    // Конфигурация продюсера
    writer := kafka.NewWriter(kafka.WriterConfig{
        Brokers:  []string{"localhost:9092"},
        Topic:    "user-events",
        Balancer: &kafka.LeastBytes{},
    })
    defer writer.Close()
    
    // Публикуем событие
    event := map[string]interface{}{
        "user_id": 42,
        "action":  "login",
        "time":    time.Now().Unix(),
    }
    
    msg := kafka.Message{
        Key:   []byte("user-42"),
        Value: []byte(jsonMarshal(event)),
    }
    
    err := writer.WriteMessages(context.Background(), msg)
    if err != nil {
        log.Fatal("Ошибка публикации:", err)
    }
    
    // Консьюмер
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers:   []string{"localhost:9092"},
        Topic:     "user-events",
        Partition: 0,
        MinBytes:  10,
        MaxBytes:  10e6,
    })
    defer reader.Close()
    
    for {
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            log.Fatal("Ошибка чтения:", err)
        }
        log.Printf("Получено сообщение: key=%s, value=%s", msg.Key, string(msg.Value))
        
        // Обработка с подтверждением
        go processMessage(msg)
    }
}

func jsonMarshal(v interface{}) []byte {
    data, _ := json.Marshal(v)
    return data
}

func processMessage(msg kafka.Message) {
    // Ваша бизнес-логика
    log.Printf("Обработка: %s", msg.Value)
}

Паттерны Event-driven архитектуры, которые спасут ваш проект

1. Event Sourcing: Храним не состояние, а историю

Вместо того чтобы хранить текущий баланс пользователя, храните все транзакции. Это дает полный аудит и возможность "откатиться" к любому моменту.

type AccountEvent struct {
    Type      string  // "deposit", "withdraw", "transfer"
    Amount    float64
    Timestamp time.Time
    AccountID string
}

// Восстановление состояния из событий
func rebuildAccount(events []AccountEvent) float64 {
    balance := 0.0
    for _, e := range events {
        switch e.Type {
        case "deposit":
            balance += e.Amount