8. CQRS и базы данных

Введение: Фундаментальное разделение моделей

Представьте крупную оптовую базу продовольствия. С заднего двора к ней непрерывно подъезжают тяжелые фуры: рабочие на вилочных погрузчиках разгружают многотонные паллеты, проверяют товарные накладные, взвешивают груз и регистрируют партии на стеллажах ответственного хранения. Здесь царят строгий учет, правила пожарной безопасности и тяжелая промышленная техника (это контур записи — Commands). А с фасада здания расположен светлый и уютный розничный супермаркет: покупатели с легкими корзинками неторопливо выбирают вымытые яблоки, сканируют штрихкоды и оплачивают покупки на терминалах самообслуживания (это контур чтения — Queries).

Если бы директор базы решил объединить эти два мира и заставил покупателей с авоськами ходить прямо между работающими погрузчиками, чтобы доставать по одному апельсину из оптовых ящиков, на складе наступил бы мгновенный паралич, а несчастные случаи стали бы неизбежностью.

CQRS (Command Query Responsibility Segregation) — это архитектурный паттерн, который разделяет операции записи (Commands) и чтения (Queries) на уровне доменной модели, кода и часто физических хранилищ данных. В отличие от классического CRUD, где одна сущность и одна таблица обслуживают оба потока, CQRS постулирует, что требования к мутациям и выборкам фундаментально различны по семантике, нагрузке и структуре данных.

Для инженера уровня Senior/Lead CQRS — это не попытка усложнить систему, а инструмент управления асимметрией требований. Он оправдан в высоконагруженных контурах с соотношением чтения к записи 100:1, в системах со сложной бизнес-логикой или когда разные потребители требуют кардинально разных представлений одних и тех же данных. Но паттерн приносит с собой распределенную сложность, задержки согласованности и новые точки отказа.

В этой статье мы разберем:

• Архитектурные границы паттерна: разделение моделей, потоки данных, уровни изоляции.
• Механику синхронизации: проекции, eventual consistency, CDC и outbox.
• Внутреннюю работу «под капотом»: влияние на паттерны IO, кэширование, аллокации и GC в Go.
• Идиоматичную реализацию проекционных воркеров, идемпотентность и обработку отказов.
• Стратегии выбора СУБД: полиглотное хранение и оптимизация под задачи чтения/записи.
• Типичные ловушки, антипаттерны и каверзные вопросы с собеседований.

Архитектура: Разделение ответственности и потоки данных

В чистом CQRS команда и запрос никогда не пересекаются в одном потоке выполнения или транзакции.

flowchart TD
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    Client["Клиентское приложение"]:::process -->|"Отправка команды"| Cmd["Command Service"]
    Cmd -->|"Валидация бизнес правил"| Bus["Шина команд"]
    Bus -->|"Изменение состояния"| WriteDB["Write Model<br>ACID СУБД"]
    WriteDB -->|"Генерация события"| Outbox["Outbox Table<br>или CDC Stream"]
    Outbox -->|"Публикация"| MQ["Очередь событий<br>Kafka или RabbitMQ"]
    MQ -->|"Подписка"| Proj["Projection Worker<br>Go горутины"]
    Proj -->|"Обновление денормализованного вида"| ReadDB["Read Model<br>PostgreSQL или ES или Redis"]
    Client -->|"HTTP или gRPC запрос"| Query["Query Service"]
  

Ключевые компоненты

1. Command Side: Принимает намерения, валидирует бизнес-правила, применяет изменения к Write Model. Гарантирует целостность и атомарность через транзакции.
2. Query Side: Отвечает на запросы чтения. Работает исключительно с Read Model, оптимизированной под конкретные сценарии (материализованные представления, денормализованные таблицы, поисковые индексы).
3. Projection Layer: Асинхронный механизм, который слушает события изменений и обновляет Read Model. Обеспечивает согласованность между двумя мирами.

Под капотом: Механика проекций и Eventual Consistency

Синхронизация между Write и Read моделями — сердце паттерна. Она всегда асинхронна, что означает eventual consistency. Данные в кэше или read-реплике могут отставать от write-модели на миллисекунды или секунды.

Паттерны доставки событий

1. Outbox Pattern: Команда пишет в основную таблицу и в служебную outbox в одной транзакции. Отдельный воркер (или CDC-инструмент типа Debezium) читает outbox, публикует события и удаляет записи. Гарантирует at-least-once.
2. Logical Replication / CDC: Чтение бинарного журнала (WAL в PostgreSQL, binlog в MySQL) на уровне СУБД. Нулевое влияние на бизнес-транзакции, но требует сложной инфраструктуры.
3. Публикация из кода: db.Exec -> kafka.Publish. Рискует потерей события при краше после коммита, но проще в реализации. Требует транзакционного outbox для надежности.

Механика согласованности и задержки

Задержка проекции (projection lag) неизбежна. Она складывается из:

• Времени публикации события в очередь (сетевой вызов, fsync брокера).
• Времени доставки до воркера (latency сети, задержки потребителей).
• Времени обработки в воркере (десериализация, бизнес-логика, запись в Read DB).
• Времени применения транзакции в Read DB (генерация WAL, обновление индексов).

Для высоконагруженных систем нормальный лаг составляет 10-500 мс. Если он растет, воркеры не справляются, и нужно горизонтально масштабировать группу потребителей или оптимизировать запросы проекции.

Идиоматичная реализация в Go: Проекционный воркер

Ниже приведен пример устойчивого воркера проекции с идемпотентностью, батчингом и корректной обработкой контекста.

package projection

import (
    "context"
    "database/sql"
    "errors"
    "fmt"
    "log"
    "time"
)

// Event представляет событие из шины
type Event struct {
    ID        string
    Type      string
    Payload   []byte
    Timestamp time.Time
}

// ProjectionWorker обрабатывает события и обновляет Read Model
type ProjectionWorker struct {
    db        *sql.DB
    batch     []Event
    batchSize int
    maxDelay  time.Duration
}

func NewProjectionWorker(db *sql.DB, batchSize int, maxDelay time.Duration) *ProjectionWorker {
    return &ProjectionWorker{
        db:        db,
        batchSize: batchSize,
        maxDelay:  maxDelay,
    }
}

// Run запускает цикл обработки. Блокирующий метод.
func (pw *ProjectionWorker) Run(ctx context.Context, eventCh <-chan Event) error {
    ticker := time.NewTicker(pw.maxDelay)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return pw.flushBatch(ctx) // Финальный сброс перед выходом
        case ev := <-eventCh:
            pw.batch = append(pw.batch, ev)
            if len(pw.batch) >= pw.batchSize {
                if err := pw.flushBatch(ctx); err != nil {
                    return fmt.Errorf("flush batch on event: %w", err)
                }
            }
        case <-ticker.C:
            if len(pw.batch) > 0 {
                if err := pw.flushBatch(ctx); err != nil {
                    return fmt.Errorf("flush batch by ticker: %w", err)
                }
            }
        }
    }
}

// flushBatch применяет накопленные события в транзакции с идемпотентностью
func (pw *ProjectionWorker) flushBatch(ctx context.Context) error {
    if len(pw.batch) == 0 {
        return nil
    }
    defer func() { pw.batch = pw.batch[:0] }() // Переиспользование слайса

    tx, err := pw.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // Подготовка идемпотентного запроса
    // Используем ON CONFLICT DO NOTHING, так как ID события уникален
    stmt, err := tx.PrepareContext(ctx, `
        INSERT INTO event_log (event_id, event_type, payload, applied_at)
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (event_id) DO NOTHING
    `)
    if err != nil {
        return fmt.Errorf("prepare stmt: %w", err)
    }
    defer stmt.Close()

    for _, ev := range pw.batch {
        _, err := stmt.ExecContext(ctx, ev.ID, ev.Type, ev.Payload, ev.Timestamp)
        if err != nil {
            // В реальном коде здесь нужна стратегия повторных попыток
            // или переход в Dead Letter Queue
            log.Printf("failed to apply event %s: %v", ev.ID, err)
            return err
        }

        // Здесь идет логика обновления конкретных Read таблиц
        // например: updateReadTableView(tx, ev.Payload)
    }

    return tx.Commit()
}

Выбор СУБД: Полиглотное хранение

Одна из главных сил CQRS — возможность использовать разные хранилища для чтения и записи.

Сложности, ловушки и антипаттерны

1. Игнорирование задержки согласованности в UI
   Пользователь создает заказ, но сразу переходит на страницу истории и не видит его. Система "сломалась"? Нет, это проекция еще не применилась.
   Решение: Оптимистичный апдейт в UI, WebSocket/SSE для пуша обновлений, или "sticky routing" (чтение всегда идет на ту же реплику, что и запись).

2. Монструозные проекции
   Попытка обновить 10 таблиц в одном событии делает транзакцию проекции долгой и блокирующей.
   Решение: Разделяйте проекции по доменным агрегатам. Пусть один воркер обновляет user_profile, другой order_stats. Это повышает параллелизм и упрощает откат.

3. Попытка сделать CQRS "на всякий случай"
   Для типового CRUD-приложения с нагрузкой 100 RPS CQRS добавляет операционные издержки без отдачи.
   Правило: Начинайте с монолитной модели. Выделяйте CQRS только когда метрики показывают, что чтение блокирует запись, или когда разные потребители требуют несовместимых схем данных.

4. Сравнение с другими экосистемами
   В Java/C# фреймворки вроде Axon или MassTransit берут на себя много магии проекций и саг. В Go принят подход явной инженерии: вы сами пишете воркеры, управляете контекстами и транзакциями. Это требует больше кода, но дает полный контроль над ресурсами, поведением при отказах и отсутствием скрытых оверхедов фреймворка.

Мониторинг и наблюдаемость

Без метрик CQRS слеп. Обязательно экспортируйте:

• projection_lag_seconds: разница между MAX(write_timestamp) и MAX(applied_at).
• events_processed_total, events_failed_total.
• read_db_query_duration_seconds, write_db_query_duration_seconds.
• batch_flush_duration_seconds, consumer_group_lag (если используется Kafka).

// Пример метрики лага
func TrackLag(ctx context.Context, dbWrite, dbRead *sql.DB) error {
    var writeTS, readTS time.Time
    // Читаем последние таймстампы...
    // Вычисляем разницу...
    // Отправляем в Prometheus
    return nil
}

Итог

CQRS — это мощный архитектурный рычаг для разделения асимметричных нагрузок. В экосистеме Go он реализуется через явные проекционные воркеры, независимые пулы соединений *sql.DB, асинхронные очереди и строгую идемпотентность. Ключевые принципы для уровня Senior/Lead:

• Разделяйте модели, когда требования к чтению и записи конфликтуют.
• Всегда проектируйте проекции как идемпотентные операции.
• Используйте батчинг и sync.Pool для снижения давления на GC и сеть.
• Мониторьте проекционный лаг как критичную метрику бизнес-здоровья.
• Не применяйте паттерн без доказанной необходимости в профилировании.

Освоив CQRS, вы получаете архитектуру, способную масштабировать чтение и запись независимо. Но как хранить состояние, если сами события становятся единственным источником истины, а транзакционные таблицы исчезают? В следующей статье мы перейдем к фундаментальному паттерну, который часто идет в паре с CQRS и меняет парадигму хранения данных: 9. Event sourcing и базы.