Лекция 219: Как связаны CQRS и Event Sourcing

Event Sourcing — это подход к хранению данных, где главным источником истины (source of truth) являются события. Вместо того чтобы сохранять текущее состояние объекта, мы сохраняем все изменения, которые когда-либо случались с ним.

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

События:
1. Пополнение: +1000 у.е.
2. Снятие: -200 у.е.
3. Пополнение: +500 у.е.

Текущее состояние:
Баланс: 1300 у.е. (вычисляется из событий)

Это позволяет не только знать "где вы сейчас", но и понять "как вы сюда попали".

Связь между CQRS и Event Sourcing

Чтобы понять, как эти два подхода связаны, давайте разберём их по полочкам.

CQRS: разделение мира на два

CQRS делит операции системы на две сферы: командный слой (Command) и слой запросов (Query). Команды изменяют состояние системы, а запросы лишь читают данные. И вот тут появляется интересная идея: а что если для изменения данных (Command) использовать события?

Event Sourcing как мощная основа CQRS

Event Sourcing идеально ложится на Command-часть CQRS. Вместо того чтобы мгновенно сохранять изменения в базу данных, вы сохраняете события, которые описывают изменения. А уже затем, на основе этих событий, обновляете состояние или создаёте новые модели для чтения данных (Query).

Смотрите, как это выглядит в действии:

1. Пользователь отправляет команду снять деньги (например, 500 у.е. со счёта).
2. Вместо того чтобы сразу обновить баланс в базе данных, система создаёт событие: MoneyWithdrawn {amount: 500}.
3. Это событие сохраняется в хранилище событий.
4. Хранилище событий позже обрабатывает это событие, обновляя модели для чтения (например, "баланс пользователя").

Таким образом, основной принцип здесь: "Сначала событие, потом последствия".

Преимущества их интеграции

Когда вы объединяете CQRS и Event Sourcing, вы получаете несколько крутых бонусов:

1. Полная история изменений. Вы можете вернуться назад во времени и понять, как объект пришёл к текущему состоянию. Это очень удобно для аудита или отладки.
2. Возможность повторного воспроизведения. Если вам нужно пересчитать текущее состояние в другой базе данных или восстановить систему после сбоя — вы просто проигрываете все события.
3. Автоматическое отделение команд и запросов. Event Sourcing органично вписывается в подход CQRS. События — это Commands, а подготовленные модели данных для запросов (Query) — это ваш читабельный результат.
4. Асинхронность. Событие можно обработать позже. Это снижает нагрузку на Command-слой и позволяет системе масштабироваться.

Реализация на практике

Давайте попробуем реализовать базовый пример с использованием CQRS вместе с Event Sourcing. Мы продолжим разработку нашего приложения для управления заказами в микросервисной архитектуре.

Предположим, наше приложение обрабатывает заказы. Нам необходимо:

• Создать заказ.
• Обновить статус заказа.
• Подготовить модель для пользователя, чтобы он мог видеть текущий статус всех своих заказов.

Шаг 1: Создание событий

Создадим Java-классы, которые представляют наши события.

// Событие создания заказа
public class OrderCreatedEvent {
    private final String orderId;
    private final String customerId;
    private final String product;

    public OrderCreatedEvent(String orderId, String customerId, String product) {
        this.orderId = orderId;
        this.customerId = customerId;
        this.product = product;
    }

    // Геттеры
}

// Событие обновления статуса заказа
public class OrderStatusUpdatedEvent {
    private final String orderId;
    private final String status;

    public OrderStatusUpdatedEvent(String orderId, String status) {
        this.orderId = orderId;
        this.status = status;
    }

    // Геттеры
}

Шаг 2: Хранилище событий

Нам нужно место, где мы будем сохранять события. Пусть для начала это будет простой список.

import java.util.ArrayList;
import java.util.List;

public class EventStore {
    private final List<Object> events = new ArrayList<>();

    public void saveEvent(Object event) {
        events.add(event);
    }

    public List<Object> getEvents() {
        return new ArrayList<>(events);
    }
}

Шаг 3: Command Handler

Обработчик команд отвечает за создание событий на основе входящих команд.

import java.util.UUID;

public class OrderCommandHandler {
    private final EventStore eventStore;

    public OrderCommandHandler(EventStore eventStore) {
        this.eventStore = eventStore;
    }

    public void handleCreateOrder(String customerId, String product) {
        String orderId = UUID.randomUUID().toString();
        OrderCreatedEvent event = new OrderCreatedEvent(orderId, customerId, product);
        eventStore.saveEvent(event);
    }

    public void handleUpdateOrderStatus(String orderId, String status) {
        OrderStatusUpdatedEvent event = new OrderStatusUpdatedEvent(orderId, status);
        eventStore.saveEvent(event);
    }
}

Шаг 4: Query Layer

Слой запросов обрабатывает события и готовит агрегированные данные для чтения.

import java.util.HashMap;
import java.util.Map;

public class OrderQueryService {
    private final Map<String, String> orderStatus = new HashMap<>();

    public void applyEvent(Object event) {
        if (event instanceof OrderCreatedEvent created) {
            orderStatus.put(created.getOrderId(), "CREATED");
        } else if (event instanceof OrderStatusUpdatedEvent updated) {
            orderStatus.put(updated.getOrderId(), updated.getStatus());
        }
    }

    public String getOrderStatus(String orderId) {
        return orderStatus.get(orderId);
    }
}

Шаг 5: Интеграция

Теперь объединим всё вместе.

public class Main {
    public static void main(String[] args) {
        EventStore eventStore = new EventStore();
        OrderCommandHandler commandHandler = new OrderCommandHandler(eventStore);
        OrderQueryService queryService = new OrderQueryService();

        // Создаём заказ
        commandHandler.handleCreateOrder("customer123", "Product A");

        // Обновляем статус заказа
        String orderId = eventStore.getEvents()
                                   .stream()
                                   .filter(event -> event instanceof OrderCreatedEvent)
                                   .map(event -> ((OrderCreatedEvent) event).getOrderId())
                                   .findFirst()
                                   .orElseThrow();

        commandHandler.handleUpdateOrderStatus(orderId, "SHIPPED");

        // Обрабатываем события в Query Service
        eventStore.getEvents().forEach(queryService::applyEvent);

        // Получаем статус заказа
        System.out.println("Статус заказа: " + queryService.getOrderStatus(orderId));
    }
}

Основные вызовы и типичные ошибки

Проблема с производительностью

При большом количестве событий проигрывание всего event log может стать узким местом. Решение? Используйте снапшоты — сохранённые промежуточные состояния.

Сложность в отладке

Событийно-ориентированные системы сложнее отлаживать из-за асинхронного характера. Хорошие логи и распределённая трассировка (например, Spring Sleuth) спасут вашу жизнь.

Консистентность данных

Важно, чтобы команды и события обрабатывались в правильной последовательности. В противном случае будет хаос.

CQRS и Event Sourcing — мощная связка для масштабируемых и отказоустойчивых микросервисов. Они позволяют разделить логику записи и чтения, сохранять полную историю действий и работать с данными асинхронно. Если их правильно настроить, они станут вашими верными союзниками, а не головной болью.