Начну с контекста. У нас order-service — сервис заказов. Стандартная CRUD-модель: заказ в PostgreSQL, статус меняется через UPDATE, аудит логируется отдельной таблицей. Всё работало, пока бизнес не сказал: «нам нужно знать полное состояние заказа на любой момент времени, не только текущее». И «нужно пересчитать комиссию за прошлый месяц для всех заказов». И «а можно откатить заказ к состоянию три дня назад?».
С CRUD это всё — боль. Аудит-таблица растёт, но по ней сложно восстанавливать состояние. Пересчёт — отдельный процесс с кастомной логикой. Откат — вообще «это невозможно».
Мы решили попробовать Event Sourcing. Храним не текущее состояние заказа, а последовательность событий: OrderCreated, ItemAdded, PaymentProcessed, OrderShipped. Текущее состояние — это просто результат применения всех событий по порядку.
Модель событий на Kotlin
Первое решение — дизайн событий. Это важнее всего. События должны быть маленькими, атомарными и неизменяемыми. Используем sealed class:
import java.time.Instant
import java.util.UUID
sealed class OrderEvent {
abstract val eventId: String
abstract val orderId: String
abstract val timestamp: Instant
abstract val version: Int
data class OrderCreated(
override val eventId: String = UUID.randomUUID().toString(),
override val orderId: String,
val userId: String,
val currency: String,
override val timestamp: Instant = Instant.now(),
override val version: Int = 1
) : OrderEvent()
data class ItemAdded(
override val eventId: String = UUID.randomUUID().toString(),
override val orderId: String,
val productId: String,
val productName: String,
val quantity: Int,
val unitPrice: BigDecimal,
override val timestamp: Instant = Instant.now(),
override val version: Int
) : OrderEvent()
data class PaymentProcessed(
override val eventId: String = UUID.randomUUID().toString(),
override val orderId: String,
val paymentId: String,
val amount: BigDecimal,
val paymentMethod: String,
override val timestamp: Instant = Instant.now(),
override val version: Int
) : OrderEvent()
data class OrderShipped(
override val eventId: String = UUID.randomUUID().toString(),
override val orderId: String,
val trackingNumber: String,
val carrier: String,
override val timestamp: Instant = Instant.now(),
override val version: Int
) : OrderEvent()
data class OrderCancelled(
override val eventId: String = UUID.randomUUID().toString(),
override val orderId: String,
val reason: String,
override val timestamp: Instant = Instant.now(),
override val version: Int
) : OrderEvent()
}Агрегат и Event Store на Kafka
Агрегат — это объект, который применяет события и выдаёт текущее состояние. Мы используем Kafka как event store: каждый агрегат имеет свой партиционный ключ (order ID), что гарантирует порядок событий.
@Component
class OrderAggregate(
private val kafkaTemplate: KafkaTemplate,
private val eventSerializer: EventSerializer
) {
private val topic = "order-events"
fun process(command: OrderCommand): OrderEvent {
val event = when (command) {
is OrderCommand.CreateOrder -> OrderEvent.OrderCreated(
orderId = UUID.randomUUID().toString(),
userId = command.userId,
currency = command.currency
)
is OrderCommand.AddItem -> OrderEvent.ItemAdded(
orderId = command.orderId,
productId = command.productId,
productName = command.productName,
quantity = command.quantity,
unitPrice = command.unitPrice,
version = getNextVersion(command.orderId)
)
is OrderCommand.ProcessPayment -> OrderEvent.PaymentProcessed(
orderId = command.orderId,
paymentId = command.paymentId,
amount = command.amount,
paymentMethod = command.paymentMethod,
version = getNextVersion(command.orderId)
)
}
// Публикуем в Kafka с ключом = orderId
kafkaTemplate.send(topic, event.orderId, eventSerializer.serialize(event))
.get(5, TimeUnit.SECONDS)
return event
}
fun rebuildState(orderId: String): OrderState {
val events = loadEvents(orderId)
return events.fold(OrderState.Initial) { state, event ->
apply(state, event)
}
}
private fun apply(state: OrderState, event: OrderEvent): OrderState {
return when (event) {
is OrderEvent.OrderCreated -> OrderState.Active(
orderId = event.orderId,
userId = event.userId,
currency = event.currency,
items = emptyList(),
status = OrderStatus.CREATED
)
is OrderEvent.ItemAdded -> {
state as OrderState.Active
state.copy(
items = state.items + OrderItem(
productId = event.productId,
name = event.productName,
quantity = event.quantity,
unitPrice = event.unitPrice
)
)
}
is OrderEvent.PaymentProcessed -> {
state as OrderState.Active
state.copy(status = OrderStatus.PAID)
}
is OrderEvent.OrderCancelled -> {
OrderState.Cancelled(event.orderId, event.reason)
}
else -> state
}
}
} Проекция и Read Model
Одна из главных идей Event Sourcing + CQRS — разделение модели записи и модели чтения. Мы пишем события в Kafka, а отдельный сервис-проекция читает их и строит материализованное представление в PostgreSQL для запросов.
@Component
class OrderProjection(private val jdbc: JdbcTemplate) {
@KafkaListener(topics = ["order-events"], groupId = "order-projection")
fun onEvent(record: ConsumerRecord) {
val event = deserializeEvent(record.value())
when (event) {
is OrderEvent.OrderCreated -> {
jdbc.update(
"""INSERT INTO orders_view (order_id, user_id, status, created_at)
VALUES (?, ?, ?, ?)
ON CONFLICT (order_id) DO NOTHING""",
event.orderId, event.userId, "CREATED", event.timestamp
)
}
is OrderEvent.ItemAdded -> {
jdbc.update(
"""INSERT INTO order_items_view
(order_id, product_id, quantity, unit_price)
VALUES (?, ?, ?, ?)""",
event.orderId, event.productId,
event.quantity, event.unitPrice
)
}
is OrderEvent.PaymentProcessed -> {
jdbc.update(
"UPDATE orders_view SET status = ?, updated_at = ? WHERE order_id = ?",
"PAID", event.timestamp, event.orderId
)
}
is OrderEvent.OrderCancelled -> {
jdbc.update(
"UPDATE orders_view SET status = ?, updated_at = ? WHERE order_id = ?",
"CANCELLED", event.timestamp, event.orderId
)
}
}
}
} Snapshotting или как не умирать от медленного replay
Заказ существует 6 месяцев, прошёл через 50 событий. Каждый раз при восстановлении состояния применяем 50 событий. Для одного заказа — ок. Для тысяч — катастрофа. Решение — snapshotting: каждые N событий сохраняем текущее состояние как слепок.
@Component
class SnapshotService(private val snapshotRepository: SnapshotRepository) {
companion object {
private const val SNAPSHOT_EVERY = 10
}
fun maybeSnapshot(
orderId: String,
events: List,
state: OrderState
) {
if (events.size % SNAPSHOT_EVERY == 0) {
snapshotRepository.save(
AggregateSnapshot(
aggregateId = orderId,
version = events.last().version,
stateJson = objectMapper.writeValueAsString(state),
createdAt = Instant.now()
)
)
}
}
fun restoreWithSnapshot(orderId: String): Pair {
val snapshot = snapshotRepository.findLatest(orderId)
val fromVersion = snapshot?.version ?: 0
val baseState = snapshot?.let {
objectMapper.readValue(it.stateJson, OrderState::class.java)
} ?: OrderState.Initial
val remaining = loadEventsFromVersion(orderId, fromVersion)
val finalState = remaining.fold(baseState) { state, event ->
apply(state, event)
}
return finalState to remaining.size
}
} Eventual Consistency — это не баг, это фича
Самый сложный аспект — eventual consistency. После записи события в Kafka проекция обновляется асинхронно. Между записью и обновлением projection проходит 50–200мс. Фронтенд может запросить заказ сразу после создания и не найти его в базе чтения.
Мы решаем это так: API возвращает событие напрямую в ответе, а для списков используем polling с exponential backoff. Да, это сложнее, чем синхронный CRUD. Но взамен мы получили полный аудит-трейл, возможность replay и бизнес-гибкость.
Когда НЕ использовать Event Sourcing
Если у вас простая CRUD-модель без сложной бизнес-логики — не используйте. Если вам не нужен аудит и replay — не используйте. Если команда не готова к смене мышления — не используйте. Event Sourcing увеличивает сложность системы на порядок, и это должно окупаться конкретными бизнес-требованиями.
Мы применили Event Sourcing только к order-service и payment-service. Остальные 10 сервисов — обычный CRUD. И это ок.
Пока нет комментариев
Будьте первым, кто оставит комментарий к этой статье