Часть 11. CQRS и Event Sourcing.
Akka Persistence.

Страничка курса: https://maxcom.github.io/scala-course-2020/

План

  1. Почему Akka Persistence?
  2. CQRS
  3. Event Sourcing
  4. Akka Persistence
  5. Снапшоты
  6. Akka Persistence Query
  7. Почему мы (почти) не используем Akka Persistence
  8. Домашнее задание

Почему Akka Persistence?

  • Знакомство с CQRS и Event Sourcing
  • Практическое применение

CQRS

command-query responsibility segregation

"Идеальная жизнь" с реляционной СУБД

  • Данные нормализованы
  • Ссылочная целостность и другие constraint'ы
  • ACID транзакции
    (Atomicity, Consistency, Isolation, Durability)

Проблема: реалиционные БД универсальны, но они не всегда могут быстро выполнить запрос.

и могут выполнить любой запрос (например полнотекстовый поиск, сложную аналитику).

В итоге появляются:

  • Производные данные в БД (предрассчитанные аггрегаты, значения из связанных таблиц, и другие варианты денормализации)
  • Внешние системы: поиск, аналитика, кеши и т.п.

Принцим CQRS - четкое разделение на две части:

  • Write side - валидирует и выполняет команды
  • Read side - только читает данные.

Каждая часть

  • Имеет свое хранилище
  • Имеет свои программные интерейсы

Как синхронизировать Read side?

  • Синхронно: триггеры и пр.
  • Асинхронно: очереди

(картинка откроется в новом окне)
© A CQRS Journey – Microsoft

"Надежная" очередь и распределенные транзакции.

Event Sourcing

"Идеальная жизнь" с реляционной СУБД

  • Данные нормализованы
  • Ссылочная целостность и другие constraint'ы
  • ACID транзакции
    (Atomicity, Consistency, Isolation, Durability)

Можно ли доверять значению в ячейке?

Например количеству денег на счете в банке?

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

Значение в БД - просто cache, вычисленный из журнала.

Еще пример: типичный компонент "enterprise" системы

  • Разнообразные "бизнес"-сущности моделированы в виде табличек БД
  • Формочки для редактирования и прочий CRUD

Этапы развития:

  • Версия 1 - кодируем модели, формочки; запускаем в работу
  • Версия 2 - удаление через флаг вместо DELETE
  • Версия 3 - отслеживаем дату последнего исправления и имя редактора
  • Версия 4 - полная история изменений

История может быть двух видов:

  • Техническая - создается автоматически; не понятна пользователю
  • "Бизнес"-история - пишется "руками" одновременно с update

Что первично?

  • текущее состояние в БД
  • журнал операций

Event Sourcing - журнал первичен.

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

События и команды (вспомним 9-ю лекцию)

  • Команда: сделай что-то. Может быть отклонена.
  • Событие: что-то произошло, это неизменяемый факт.

Что дает Event Sourcing:

  • Полную "честную" историю изменений
  • Запись append-only: хорошо для многих видов хранилищ
  • Возможность записать дополнительные сведения

Есть две проблемы:

  • Истории может быть слишком много
  • Не понятно как быстро читать и делать запросы

Event Sourcing + CQRS

  • Write Side валидирует команды и пишет события
  • Read Side - consumer к журналу событий

Read Side

  • Может быть несколько
  • Можно перестроить
  • Можно поменять схему

Akka Persistence

PersistentActor - актор + журнал событий.

При рестарте восстанавливает состояние из журнала.

build.sbt 


// версии всех akka-* должны быть одинаковые
libraryDependencies += 
  "com.typesafe.akka" %% "akka-persistence-typed" % "2.6.4"
libraryDependencies += // или другой backend Slf4j
  "ch.qos.logback" % "logback-classic" % "1.2.3"

Команда - запрос актору на выполнение действия.

Команда проходит валидацию, сохраняет событие и выполняет побочный эффект.

Пример - вспомним CounterActor: 


object CounterActor {
  sealed trait CommandProtocol
  case class Get(replyTo: ActorRef[Int]) extends CommandProtocol
  case class IncrWhen(amount: Int, pred: Int => Boolean)
    extends CommandProtocol

  ...
}
					

def behavior(counter: Int = 0): Behavior[CommandProtocol] =
  Behaviors.receiveMessage[CommandProtocol] {
    case Get(replyTo) =>
      replyTo ! counter
      Behaviors.same
    case IncrWhen(amount, pred) =>
      if (pred(counter + amount)) {
        behavior(counter + amount)
      } else {
        Behaviors.same
      }
  }
					

def behavior: Behavior[CommandProtocol] = 
  EventSourcedBehavior[CommandProtocol, EventProtocol, Int](
    persistenceId = PersistenceId.ofUniqueId("myCounter"),
    emptyState = 0,
    // (Int, CommandProtocol) => Effect[EventProtocol, Int]
    commandHandler = ???,    
    // (Int, EventProtocol) => Int
    eventHandler = ???)
  • persistenceId - идентификатор в хранилище
  • commandHandler - преобразует комманды в эффекты
  • eventHandler - обновляет state по событиям

Основные эффекты:

  • none/reply - для запросов
  • persist - запись нового события

См. еще эффекты в документации

Внешний побочный эффект: 


Effect.persist(EmailSent(data)).thenRun(sendEmail())
					

object CounterActor {
  ...

  sealed trait EventProtocol
  case class Incr(amount: Int) extends EventProtocol
  
  ...
}
					
val commandHandler: 
    (Int, CommandProtocol) => Effect[EventProtocol, Int] = { 
  (counter, command) =>
    command match {
      case Get(replyTo) =>
        Effect.reply(replyTo)(counter)
      case IncrWhen(amount, pred) =>
        if (pred(counter + amount)) {
          Effect.persist(Incr(amount))
        } else {
          Effect.none
        }
    }
}

val eventHandler: (Int, EventProtocol) => Int = { 
  (state, event) =>
    event match {
      case Incr(amount) => state + amount
    }
}

При восстановлении работает eventHandler

В нем не должно быть побочных эффектов

Для запуска не хватает настроек Storage и Serialization

Нужен плагин для storage. Обычно используют Cassandra, для отладки проще LevelDB

build.sbt


// плагин встроен, нужна только сама LevelDB
libraryDependencies += 
  "org.fusesource.leveldbjni" % "leveldbjni-all" % "1.8"

src/main/resources/application.conf 


akka.persistence.journal.plugin = 
  "akka.persistence.journal.leveldb"
akka.persistence.snapshot-store.plugin = 
  "akka.persistence.snapshot-store.local"

akka.persistence.journal.leveldb.dir = 
  "target/example/journal"
akka.persistence.snapshot-store.local.dir = 
  "target/example/snapshots"
// snapshot потом пригодится

Настроим сериализацию

build.sbt


libraryDependencies += 
  "com.typesafe.akka" %% "akka-serialization-jackson" % "2.6.4"
					

trait MySerializable

sealed trait EventProtocol extends MySerializable

src/main/resources/application.conf 


akka.actor {
  serialization-bindings {
    "MySerializable" = jackson-json
  }
}
					

object ActorDemo extends App {
  implicit val actorSystem: ActorSystem[CommandProtocol] = 
    ActorSystem(CounterActor.behavior, "counter")
  implicit val timeout: Timeout = 1.minute

  actorSystem ! IncrWhen(10, _ => true)

  println(Await.result(actorSystem ? Get, timeout.duration))
}

при перезапусках значение восстанавливается

Снапшоты

Проблема: много событий и восстановление слишком долгое.

Решение - периодически записывать снапшоты.

Включаем снапшоты 


def behavior: Behavior[CommandProtocol] =
  EventSourcedBehavior[CommandProtocol, EventProtocol, Int](
      persistenceId = PersistenceId.ofUniqueId("myCounter"),
      emptyState = 0,
      commandHandler = commandHandler,
      eventHandler = eventHandler)
    .withRetention(
      RetentionCriteria.snapshotEvery(
        numberOfEvents = 10, 
	keepNSnapshots = 2)
    .withDeleteEventsOnSnapshot)
					

(0 to 20).foreach { _ =>
  actorSystem ! IncrWhen(10, _ => true)
}

запустите и посмотрите debug log

Akka Persistence Query

Event sourcing + CQRS:

  • Write Side: пишет события в журнал
  • Read Side: дополнительные consumer'ы журнала

build.sbt 


// версии всех akka-* должны быть одинаковые
libraryDependencies += 
  "com.typesafe.akka" %% "akka-persistence-query" % "2.6.4"

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


// создаем один раз на приложение
val queries: LeveldbReadJournal = 
  PersistenceQuery(system).
    readJournalFor[LeveldbReadJournal]
      (LeveldbReadJournal.Identifier)

// получаем новый поток событий
val events: Source[EventEnvelope, NotUsed] =
  queries.currentEventsByPersistenceId("myCounter", 
                                       fromSequenceNr = 0)

events.runForeach(println) // запустите и посмотрите

Что можно делать с потоком событий (примеры)?

  • Строить актор с выдержкой из текущего состояния
  • Перекачивать во внешние хранилище (индекс, БД, т.п.)
  • Считать статистику над потоком событий
  • и др.

Есть механизм чтения событий
из нескольких акторов по "тегу".

Read side обновляется асинхронно.

Иногда нужно прочитать результат после записи.

Можно при записи отдать sequenceNq пользователю.

После реализовать логику ожидания нужного номера в view.

Почему мы (почти) не используем Akka Persistence

Реляционная БД удобнее:

  • Легче отлаживать (хотя json тоже можно смотреть)
  • Можно "пофиксить" запросом

Домашнее задание

Добавить persistence к актору который загружает ленту.

  • Сохраняем все обновления ленты на диск
  • Стартуем с сохраненной лентой
  • Периодически пишем снапшоты

Напоминаю: