Notifications via Timeouts. Як створити сервіс нотифікацій за допомогою NServiceBus

Підписуйтеся на Telegram-канал «DOU #tech», щоб не пропустити нові технічні статті

Всім привіт. Мене звати Олександр Шпортько. Я працюю .NET-розробником в компанії ITERA. Основним доменом, з яким я працюю є банківський сектор. Це, як ви розумієте, і багато операцій з трансфером грошей, і кредити/ депозити та, звичайно, нотифікації, і ще раз нотифікації, за допомогою яких користувачі отримують повідомлення про стан своїх грошових транзакцій.

Тож в цій статті я хотів би поділитися своїм досвідом зі створення саме сервісу нотифікацій за допомогою NServiceBus, але в трішки іншій та не менш цікавій сфері.

В планах у нас пірнути в проблематику цього питання, ознайомитися з тим, що таке NServiceBus та які він має особливості, а потім розглянути солюшен, який містить реалізацію сповіщень.

А наостанок, подивимось, як працюють відкладені сповіщення та підсумуємо усе написане.

Intro

На сьогодні існує багато сервісів, які виконують роль нотифікацій в залежності від подій в самій системі або від дій користувача. Серед основних можна відзначити музичні, ігрові, стримінгові сервіси, нотифікації у вигляді sms чи email повідомлень й таке інше.

А чи не спадало вам на думку, що всі ці, на перший погляд, елементарні меседжі та інформаційні сповіщення можуть бути як дуже прості, так і досить складні, що залежать від багатьох умов. Наприклад, що відбувається, коли ми підписуємось на Netflix, вказуючи вигідний для вас план зі строком підписки? Як система вміє надіслати користувачеві нагадування, що змінився тариф, закінчується строк підписки і таке інше.

Певно, у вас все ж таки є думки з приводу цього, як-то тісна взаємодія між мікросервісами, база даних, шедулєри. Так. Але не зовсім. А може навіть простіше, ніж ми уявляємо. Отже, перейдімо безпосередньо до сабжу.

Розглянемо приклад підписки на сервіс Netflix. Проєкт реалізує деякий функціонал зі створення користувачів, вибір тарифних планів та підписку, а також нотифікації (як одночасні, так і відкладені у часі). По суті це WEB API (.NET Core) солюшен з використанням NServiceBus — масштабована платформа для обміну сповіщеннями.

Базою даних для нашої Api ми використаємо MS SQL. NServiceBus для своїх потреб також буде використовувати MS SQL: чергу з меседжів, що надходять (SQL Server transport), а також для збереження стану саги (persistence) на час обробки цих меседжів. Чому саме MS SQL? По-перше, багато компаній (DELL, Ebay, Fujitsu та інші), що мають стек з Microsoft продуктів, вже розуміють як підтримувати їх.

А по-друге, NServiceBus автоматично створить всі необхідні таблиці для обробки сповіщень (далі ми розглянемо такі поняття як Saga та Timeouts), а це позбавить нас зайвої роботи з розгортання проєкту.

NServiceBus

NServiceBus — це потужний фреймворк, який допомагає розробнику створювати розподільні системи, які є більш надійними, гнучкими, масштабованими та легшими для підтримки та оновлення. Це все досягається завдяки тому, що NServiceBus пропонує інфраструктурні рішення прямо з коробки та дозволяє сфокусуватися на бізнес-логіці солюшена і не відволікатися, а тим паче, витрачати додатковий час та гроші бізнесу, на побудову message-driven сервісів с нуля. Подивіться самі:

• вбудований exception handling інструментарій (моніторинг та детальний аналіз);
• механізм автоматичних повторних спроб (retry) які можна налаштувати під потреби бізнесу (одразу, через деякі проміжки часу);
• гарантована доставка сповіщень (events, commands): ви можете бути впевнені в їх доставці адресату завдяки черзі невдалих (через exception) сповіщень, які згодом можуть бути обробленні та відправлені знов;
• містить транспортну абстракцію для основних технологій масового обслуговування (MSMQ, Azure Service Bus, SQL Server, RabbitMQ тощо);
• вбудована імплементація патерну publish-subscriber;
• асинхронний messaging;
• Оркестрування бізнес-процесів — оркестратор (Saga) повідомляє учасникам процесу, які локальні транзакції (single unit of logic work) виконувати.

Одною з цікавих особливостей NServiceBus у нашому прикладі є long-running процеси, які реалізовуються за допомогою Saga, та оркестратор, що визначає поведінку та впроваджує бізнес-правила. Одним з прикладів таких «довгих» процесів може бути обробка великої кількості email з подальшою відправкою листів. Ми могли використовувати batch обробники, але вони можуть тривати дуже довго і, можливо, нам би довелося запускати їх кожного дня. А добре було б лише тоді, коли це нам потрібно та з мінімальними зусиллями від розробника. Тому, нам на допомогу приходять Saga.

Як саме працює Saga? Насамперед відзначимо, що сага — це послідовність локальних операцій (транзакцій), де кожна така локальна транзакція виконує певну логіку, наприклад оновлює базу даних і публікує повідомлення для запуску наступної локальної транзакції в сазі. Саги можна розпочати одним або кількома повідомленнями. Щойно повідомлення надходить і визначається, як початкове, створюється новий екземпляр саги, і повідомлення надсилається в цей екземпляр саги. Вона може взаємодіяти з повідомленням, як звичайний обробник повідомлень, але може зберігати певний стан своїх даних. Відразу ж після завершення повідомлення стан саги зберігається в постійному сховищі (у БД). Розпочата сага має унікальний ідентифікатор кореляції (CorrelationId), який додається до кожного повідомлення, надісланого з саги. Це дозволяє зв’язати всі вхідні повідомлення з правильним екземпляром саги. Як тільки обробка повідомлення завершена з точки зору бізнес-потреб, її треба завершити (MarkAsComplete).

Ще однією особливістю NServiceBus є те, що він має вбудовану імплементацію патерну publish/subscribe. Тобто деякий сервіс може опублікувати івент і не підозрювати, ким він буде оброблений, а ось сторона, що приймає, точно буде чекати на цей івент, а згодом його оброблювати. Це дуже розв’язує (decouple) сервіси між собою, що дозволяє обробляти сповіщення асинхронно та незалежно одне від одного.

NServiceBus робить додатки слабо зв’язаними, спілкуючись асинхронно, що також робить зв’язок надійнішим, оскільки дві програми не повинні працювати одночасно. На додачу, якщо частина системи виходить з ладу, дані не втрачаються.

Звичайно, ми можемо протиставити тому всьому RPC технології, мовляв, то є швидше та простіше. Але за допомогою NServiceBus ми отримуємо не крихку, але досить складну систему, де одна частина системи сповільнюється, впливаючи на інші частини системи, зрештою призводячи до збою всієї системи, а надійний продукт, який скеровує розробників писати код, який є надійним у production середовищах.

Не слід забувати той факт, що NServiceBus дає нам переваги eventual consistency, через асинхронність меседжів. Кожен отримувач обробляє дане повідомлення щоразу, коли йому трапляється отримати це повідомлення з черги. Зворотна сторона медалі в цьому аспекті — це те, що система стане консистентною тільки через певний проміжок часу, що може не задовольнити деякі бізнес-потреби.

Implementation

Як ми можемо побачити, проєкт дуже нагадує за своєю архітектурою багатошарову onion-подібну структуру, де кожен шар максимально відображає те, за що він відповідає. Схематично наш солюшен можна зобразити наступним чином:

Структура солюшена розділена на логічні одиниці, що між собою не міксуються, але тісно взаємодіють:

Пояснимо кожен із них відповідно до нашого дизайну.

Business — прошарок Domain, який містить елементи, що максимально зображають бізнесову модель. Доменні модельки (user, plan, subscription), інтерфейси та сервіси Service, що виконують бізнес-задачі.

Resources — даний Repository прошарок солюшена містить об’єкти та сервіси для роботи з базою даних (запис, отримання та видалення даних), сервіс формування повідомлень на пошту.

DbUp — мабуть, найпростіший проєкт за своєю структурою, тому що містить скрипти для створення нашої бази даних (створення табличок та наповнення).

Shared — константи, enums, команди для комунікації між Api та NServiceBus-ом.

Api — проєкт, який містить основні компоненти рівня Api, як-то контролери (UserController, PlanController, SubscriptionController), моделі запитів (query, request), мапери (mappers).

NServiceBus — однойменний проєкт, відповідає за прийом команд від інших сервісів (в нашому випадку від Api) та обробку цих команд із подальшим формуванням нотифікацій.

Контролери:

• UserController (створення користувачів, отримання всіх користувачів та по id, активація/деактивація користувача).
• PlanController (створення плану підписки, отримання всіх планів, зміна прайсу).
• SubscriptionController (створення підписок із датою її закінчення, видалення підписки по id, отримання списку підписок).

Дії, на які ми будемо відправляти нотифікації:

• Активація/деактивація користувача.
• Зміна прайсу (ціни на підписку).
• Зміна дати дії підписки.
• Нагадування про закінчення підписки через деякий час.

Всі операції будемо виконувати за допомогою Postman та вже створеної колекції.

База даних MS SQL.

Загалом, флоу маємо наступний:

Saga

Щоб відправити повідомлення та обробити його за певною бізнес-логікою, нам потрібна саме Saga. Основна одиниця NServiceBus, яка відповідає за отримання сповіщень від інших сервісів та направлення їх на оброку до відповідних handler-ів.

В нашому випадку це виглядає наступним чином NotificationSaga.cs

Сага зберігає стан меседжів у вигляді даних самої саги (в нашому випадку в БД MS SQL в таблицю з найменуванням «...NoficationSaga»). Саме тому наш клас, що буде оброблювати інформацію, що надходить від інших сервісів, а це Api-проєкт, успадковується від Saga<T> класу. В якості T має бути клас, який має окреме поле CorreleationId, за допомогою якого буде відбуватися мапінг та пошук відповідної саги у БД (ConfigureHowToFindSaga).

select * from SagaDataTable where CorrelationId = @Id — таким запитом сервіс-бас шукає збережену сагу в базі даних.

Річ у тім, що ми можемо зробити багато таких саг і нам потрібно вказати NServiceBus, яка саме нас цікавить сага при обробці івенту, що надходить.

Сага має життєвий цикл, тому нам потрібно попіклуватися про те, щоб вона коректно розпочала своє функціонування та коректно її завершила. За допомогою IAmStartedByMessages<T> ми даємо інструкцію NServiceBus створити сагу і розпочати обробку однієї з вхідної команди Т, де Т — це UserNotificationCommand, PlanNotificationCommand, SubscriptionNotificationCommand. Як виглядають хендлери для них можна глянути за посиланнями: UserNotificationCommand, PlanNotificationCommand, SubscriptionNotificationCommand.

Завершиться сага в нас при обробці SendEmailResponse, коли ми вже відправили наше повідомлення у вигляді email користувачеві. Або коли ми вирішили, що відпрацювали всі таймаути. Обов’язково вказати метод MarkAsComplete() - SendEmailResponse.

Notifications via Timeouts

Ви, мабуть, вже стикалися з ситуацією, коли сучасні розумно побудовані сервіси підписок (музичні, ігрові, стримінгові та ін.) мають властивість нагадувати нам, що підписка скоро закінчиться, або що залишається декілька днів до зміни тарифу і таке інше.

Іноді нагадування можуть бути зроблені з деяким проміжком часу. Таймаути можуть виконати свою логіку як на заданий час (не раніше, ніж встановлений час в майбутньому) або з затримкою у часі (відкладені меседжі). В нашому випадку ми будемо використовувати таймаут як будильник. А точніше, як декілька будильників на задані часи.

Тож, як ми створюємо таймаути. Це дуже просто. Нам потрібен власний кастомний клас, SubscriptionTimeout.cs, вказати сазі, що вона буде оброблювати наш таймаут (IHandleTimeouts<SubscriptionTimeout>) та прописати handler для цього таймауту Handle(SubscriptionNotificationCommand message, IMessageHandlerContext context).

При цьому, треба вказати дуже цікаву особливість таймаутів. Річ у тім, що зазвичай, коли сага ловить якийсь меседж/команду, вона зберігає свій ідентифікатор у БД (в нашому випадку — це MS SQL) в табличці NServiceBus_NotificationSaga

і тримає цей запис там до тих пір, допоки вся бізнес-логіка, що прописана в хендлерах, не буде виконана і не настане команда MarkAsComplete(). Післе цієї команди сага знищується.

А коли ми створюємо таймаут, тоді ми повинні подбати про те, щоб сага, яка «зловила» нашу команду, не знищувалась. Тобто прописати умову її завершення.

Виклик таймауту відтворюємо наступним чином. Коли користувач активує свою підписку, ми заводимо таймаут, орієнтований на дату її закінчення ExpirationDate. Але спочатку ставимо очікування відправки повідомлення про завершення підписки за 10 днів до її завершення, потім — за 3 дні і за день. При цьому ставимо прапорець IsLastNotification = true для того, щоб хендлер таймауту зрозумів, що потрібно завершувати сагу.

При цьому в БД ми будемо спостерігати наступну картину. В таблиці NServiceBus_NotificationSaga:

В таблиці NServiceBus.Delayed (таймаути):

У ході обробки кожного таймауту користувач буде отримувати сповіщення, а відповідний запис у БД буде видалятися.

А на останок, коли нарешті користувач проігнорує всі наше повідомлення, ми автоматично видаляємо його підписку, тим самим деактивуємо її за допомогою DELETE /Subscription/{id}. І також надсилаємо сповіщення про це.

Також необхідно зазначити, що у разі якогось іксепшену (наприклад, наша БД лягла та не відповідає) інфраструктура NServiceBus самостійно буде намагатись повторити відпрацювання саги через певний проміжок часу. І це дуже зручно, тому що нам не потрібно прописувати цей дуже великий кусок коду.

Conclusions

Отже, ми розглянули дуже гнучке та потужне рішення NServiceBus від Particular Software для обміну повідомленнями, а також керуванням обробкою цих повідомлень. NServiceBus дозволяє розгорнути всю необхідну інфраструктуру з мінімальними зусиллями.

Основним механізмом обробки повідомлень можна вважати Saga. Вона приймає повідомлення від інших сервісів, делегує логіку виконання хендлерам, які своєю чергою можуть виконувати такі дії як: запис у БД, відправку нотифікацій, виклик сторонніх API тощо. І якщо нам потрібно побудувати систему з відкладеними нотифікаціями, то ми робимо це на базі timeouts, які дають можливість уникнути створення зайвих шедулєрів на базі тої чи іншої ОС на сервері.

Крім цього, ми можемо через спеціальний UI інструмент ServicePulse (який йде разом із NServiceBus) моніторити іксепшени та меседжі, які оброблюють саги.

Звісно, що колись настає той час, коли систему треба масштабувати, якщо змінилися або бізнес-потреби, або виросла кількість інформації, що оброблюється. В такому випадку ми не можемо оперувати кількісними значеннями, наприклад, більш як 1 мільйон меседжів NServiceBus опрацювати не зможе.

Масштабування системи може залежати як від оптимізації самого коду, що оброблює дані, так і від архітектурних та інфраструктурних рішень. Щодо цього є дуже цікаві приклади: Scaling with Asynchronous Messaging та реальний кейс з використанням NServiceBus в обробці запитів фотографій у Spotlight від Dylan Beattie — Media publishing workflows using NServiceBus • Particular Software.

Матеріали, що використовувалися:

• Learning NServiceBus: Boike, David: Amazon.com: Books
• Particular Developer Portal