Sergey-nsk Jan 9 2019 at 13:44

Снимки событий в Axonframework 3, улучшаем производительность

5 min

2.4K

Обзор фреймврока Axonframework

Axonframework это фреймфорк реализующий несколько принципов и паттернов проектирования такие как:

CQRS – разделяет обработку запросов на чтение и запись данных
Event Sourcing – это когда состояние приложения хранится как цепочка событий
DDD Aggregate – доменный объект (domain object) который хранит состояние

Один из недостатков хранения конечного состояния приложения в виде цепочки событий – это количество хранимых и обрабатываемых событий. К счастью, Axonframework позволяет создавать снимок событий (snapshot event), который содержит в себе результат нескольких событий (domain event).

Снимки событий

Снимок событий (snapshot event) – это результирующие значения нескольких событий (domain event). Это позволяет быстрее воссоздавать состояние Агрегата (Aggregate). Важно понимать, что снимок создаётся из событий которые применялись для конкретного Агрегата с уникальным идентификатором.

Например (рис.1), зададим в конфигурации создание снимка на каждые два события (порог = 2 — для наглядности примера). В таком случае, когда два события изменят состояние Агрегата, то создастся один снимок c результирующими значениями предыдущих двух событий.

Рис 1. Снимок двух событий. (порог=2)

Рассмотрим пример посложнее(рис.2), в конфигурации также указан порог равный 2, чтобы снимок создавался каждые два события. Когда 2 события изменят состояние Агрегата, то создастся один снимок. Далее другие 2 события изменяют состояние Агрегата и новый снимок не создаётся, а обновляется уже существующий.

Рис.2 Результат цепочки событий в одном снимке (порог=2)

Производительность

С одной стороны когда в приложении накапливается длинная цепочка событий, то требуется время для чтения и обработки большого числа событий для воссоздания состояния Агрегата. С другой стороны, если создавать снимок, то состояние Агрегата будет воссоздано быстро, но потребуется время на создание снимка. Необходимо находить баланс между этими двумя ситуациями.

Рис.3 Производительность без создания снимка

Рис.4 Производительность с созданием снимка (порог = 3)

По умолчанию, снимок создаётся в потоке который вызвал метод scheduleSnapshot(). Такая настройка не рекомендуется для боевой среды (см рис.4/запись).

Ниже приведём пример кода с применением ThreadPoolExecutor(...) который предоставит отдельный поток для создания снимка. В таком случае, наш клиент не заметит замедления в работе приложения и выделенное время на создание снимка.

Код

Для активации создания снимков требуется внести небольшие изменения в код приложения. В аннотации Агрегата указывается имя репозитория которое используется в коде конфигурационного класса. В конфигурационном классе указывается порог для создания снимков, способ создания снимков, репозитории и т.п.

AxonConfig.java

@Autowired
private EventStore eventStore;

@Bean
public SpringAggregateSnapshotterFactoryBean springAggregateSnapshotterFactoryBean() {
   return new SpringAggregateSnapshotterFactoryBean();
}
@Bean
public SpringAggregateSnapshotter snapshotter(ParameterResolverFactory parameterResolverFactory, EventStore eventStore, TransactionManager transactionManager) {
   Executor executor = Executors.newFixedThreadPool(10);
   return new SpringAggregateSnapshotter(eventStore, parameterResolverFactory, executor, transactionManager);
}

@Bean("reservationRepository")
public EventSourcingRepository<Reservation> reservationRepository(Snapshotter snapshotter, ParameterResolverFactory parameterResolverFactory) {
   return new EventSourcingRepository<Reservation>(reservationAggregateFactory(), eventStore, parameterResolverFactory, new EventCountSnapshotTriggerDefinition(snapshotter, 50));
}

@Bean(name = "reservationAggregateFactory")
public AggregateFactory<Reservation> reservationAggregateFactory() {
   SpringPrototypeAggregateFactory<Reservation> aggregateFactory = 
   new SpringPrototypeAggregateFactory<>();
   aggregateFactory.setPrototypeBeanName("reservation");
   return aggregateFactory;
}

Reservation.java

@Aggregate(repository = "reservationRepository")
public class Reservation {
	//…
}

Стоит отметить, что в ветке обсуждения Google Groups содержатся полезные примеры кода и обсуждения.

Выбор порогового значения для создания снимков

5.1. Теоретический путь

Посчитаем количество событий которые могут применяться к Агрегату в EventListener классе. Затем теоретически оценим среднее количество событий применяемых к Агрегату в типичной ситуации и значение несколько меньше этого установим в качестве порогового для создания снимков. Так можно поступить если приложение только создано и нет реальных данных для анализа.

5.2. Практический путь

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

> docker exec -it <container-id> mongo 
> show dbs
admin          	0.000GB
axonframework	0.000GB
local          	0.000GB

> use axonframework
switched to db axonframework

> show collections
domainevents
sagas
snapshotevents

> db.domainevents.findOne()

{
 “_id” : ObjectId(“5bb1dc8d4446d63bcc765feb”),
 “aggregateIdentifier” : “b1e320d5–58aa-4b9b-a667-aa724900592f”,
 “type” : “Reservation”,
 “sequenceNumber” : NumberLong(0),
 “serializedPayload” : “<com.example.ReservationStarted><reservationIdentifier>b1e320d5–58aa-4b9b-a667-aa724900592f</reservationIdentifier><duration resolves-to=\”java.time.Ser\”><byte>1</byte><long>2400</long><int>0</int></duration></com.example.ReservationStarted>”,
 “timestamp” : “2018–10–01T08:36:29.434Z”,
 “payloadType” : “com.example.ReservationStarted”,
 “payloadRevision” : null,
 “serializedMetaData” : “<meta-data><entry><string>traceId</string><string>b090b86a-ec89–484b-ae9f-e4fa0f9bcd39</string></entry><entry><string>correlationId</string><string>b090b86a-ec89–484b-ae9f-e4fa0f9bcd39</string></entry></meta-data>”,
 “eventIdentifier” : “f324f021–50b4–4e91–84d0-f8c4425f3eb9”
}

Каждое хранящееся событие содержит поле aggregateIdentifier, по которому посчитаем количество событий примененных к каждому Агрегату простым запросом:

db.domainevents.aggregate([ 
    {$group: {_id: "$aggregateIdentifier", count: {$sum: 1} } },
    {$sort : {count : -1} }
]);

{ "_id" : "0d84afd1-f199-45c8-b50e-7d9ebfa4c8fb", "count" : 136 }
{ "_id" : "49de7c32-38ea-435a-b837-ccdb61ec0baa", "count" : 136 }
{ "_id" : "12957b0b-af05-47c4-a3d8-968b75cf9ffb", "count" : 136 }
{ "_id" : "97a24559-ee3a-43e7-a6be-1eb6840b662a", "count" : 132 }
{ "_id" : "b6aeb1af-0620-4b02-8de3-c2446c2f7d83", "count" : 132 }
{ "_id" : "b385aaf4-3338-489f-8d1b-4600d5e088b9", "count" : 132 }
{ "_id" : "5970327f-9551-4945-94e9-3844c0cd3543", "count" : 132 }
...
{ "_id" : "0182239h-3948-3334-98t5-9643j4ld8346", "count" : 1 }

Пороговое значение для создания снимков можно выбрать меньше среднего чтобы снимки создавались эффективно. В данном случае значение 50 вполне подойдёт.

Проверка активации снимков

> mongo
> show dbs
admin          	0.000GB
axonframework	0.000GB
local          	0.000GB

> use axonframework
> show collections
domainevents
sagas
snapshotevents

> db.domainevents.count()
515
> db.snapshotevents.count()
7

Если коллекция snapshotevents не пустая и содержит в себе снимки, то создание снимков активировано успешно.

Другие возможности создания снимков

В документации упоминаются и другие вариации по активации создания снимков, например:

число событий созданных с момента последнего снимка превысило пороговое значение
время на инициализацию Агрегата истекло
временная задержка и т.д. и т.п.

Tags:

Hubs:

Java