Как воспроизводятся отсутствующие события?

Question

Я пытаюсь узнать больше о CQRS и Event Sourcing (хранилище событий).

Насколько я понимаю, очередь сообщений / шина обычно не используются в этом сценарии - шина сообщений может использоваться для облегчениясвязь между микросервисами, однако она обычно не используется специально для CQRS.Однако, как я вижу это сейчас, шина сообщений была бы очень полезна, гарантируя, что модель чтения в конечном итоге синхронизируется, и, следовательно, возможна согласованность, например, когда сервер, на котором размещена база данных модели чтения, снова подключается.

Я понимаю, что возможное соответствие часто приемлемо с CQRS.Мой вопроскак сторона чтения узнает, что она не синхронизирована со стороной записи?Например, предположим, что в обычном дне в хранилище событий создается 2 000 000 событий, а 1 999 050 также записываются в хранилище чтения.Оставшиеся 950 событий не записываются из-за ошибки в программном обеспечении или из-за того, что сервер, на котором размещена модель чтения, находится в автономном режиме в течение нескольких секунд.Как возможная последовательность работает здесь?Откуда приложение знает, как воспроизвести события 950, которые отсутствуют в конце дня, или события x, которые были пропущены из-за простоя десять минут назад?

Я читал здесь вопросы за последнюю неделюили около того, в которых говорится о сообщениях, воспроизводимых из хранилища событий, например: CQRS - воспроизведение событий для стороны чтения , однако никто не говорит о том, как это делается.Нужно ли настраивать запланированное задание, которое запускается один раз в день и воспроизводит все события, созданные с момента последнего выполнения запланированного задания?Есть ли более элегантный подход?

Answer 1

Я использовал два подхода в моих проектах, в зависимости от требований:

1. Синхронный, внутрипроцессный Readmodels.После того, как события сохраняются, в том же времени существования запроса, в том же процессе, Readmodels получают эти события.В случае сбоя Readmodel (ошибка или перехватываемая ошибка / исключение) ошибка записывается в журнал, и что Readmodel просто пропускается, а следующий Readmodel получает события и так далее.Затем следуйте Сагам, которые могут генерировать команды, которые генерируют больше событий, и цикл повторяется.

Я использую этот подход, когда влияние сбоя Readmodel является приемлемым для бизнеса, когда готовностьДанные Readmodel важнее, чем риск сбоя.Например, они хотели, чтобы данные были немедленно доступны в пользовательском интерфейсе.

Журнал ошибок должен быть легко доступен на некоторой административной панели, чтобы кто-то мог просмотреть его в случае, если клиент сообщает о несоответствии между командами записи / чтения и чтения / запроса..

Это также работает, если у вас есть свои Readmodels, связанные друг с другом, то есть одному Readmodel нужны данные из другого канонического Readmodel.Хотя это кажется плохим, это не так, это всегда зависит.Бывают случаи, когда вы торгуете дублированием кода / логики программы обновления с упругостью.

Асинхронный, в другом процессе Средство чтения readmodel.Это используется, когда я использую полное разделение Readmodel от других Readmodels, когда сбой Readmodel не приведет к снижению всей стороны чтения;или когда Readmodel нужен другой язык, отличный от монолита.В основном это микросервис.Когда что-то плохое происходит внутри Readmodel, необходимо, чтобы какой-то авторитетный компонент более высокого уровня был уведомлен, то есть администратор был уведомлен по электронной почте, SMS или другим способом.

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

При любом подходе модели чтения должны легко перестраиваться.

КакВы бы выбрали между подходом тяги и подходом толчка?Если бы вы использовали очередь сообщений с push (событиями)

Я предпочитаю подход на основе извлечения, потому что:

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

Бывают случаи, когда я выбираю очередь сообщений:

• , вам нужно, чтобы события были доступны, даже если хранилище событий не
• вам нужны конкурентные / параллельные потребители
• вы не хотите отслеживать, какие сообщения вы потребляете;по мере их использования они автоматически удаляются из очереди

Answer 2

Приложение не знает, что оно не обработало некоторые события из-за ошибки.

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

В случае ошибки в проекции / инфраструктуре, которая привела к тому, что определенный прогноз является недействительным, вам может потребоваться перестроить этот прогноз,В большинстве случаев это будет ручное вмешательство, при котором контрольная точка проекции сбрасывается до 0 (начало времени), поэтому проекция извлекает все события из хранилища событий с нуля и повторно обрабатывает их все.

Answer 3

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

Каждый проектор имеет отслеживание положения, где он находится вдоль порядкового номера.Проекции похожи на логические очереди.

Вы можете очистить данные проекции и сбросить позицию обратно на 0, и она должна быть восстановлена.

В вашем случае проекция не удаласьпо какой-то причине, например, когда сервер переходит в автономный режим, на позицию 1 999 050, но при повторном запуске сервера он продолжит работу с этого момента.

Answer 4

Этот доклад от Грега Янга может помочь.

Как приложение узнает, что можно воспроизвести 950 событий, отсутствующих в конце дня, или события x, которыебыли пропущены из-за простоя десять минут назад?

Итак, здесь есть два разных подхода.

Один из них, возможно, проще, чем вы ожидаете - каждый раз, когда вам нужно перестроить модель чтенияпросто начните с события 0 в потоке.

Да, масштаб в конечном итоге отстой, так что вы не захотите, чтобы это была ваша первая стратегия.Но обратите внимание, что это работает.

Для обновлений с не столь смущающими свойствами масштабирования обычная идея состоит в том, что модель чтения отслеживает метаданные о положении потока, использованные для построения предыдущей модели.Таким образом, запрос из модели чтения становится «Что произошло после события № 1,999,050»?

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

EventStore.ReadStreamEventsForwardAsync(stream, 1999050, 100, false)