Челленджі в імплементації розподіленої системи відправки повідомлень із використанням передових підходів у .NET

Привіт усім. Мене звати Владислав Фордак. У цій доповіді ми розглянемо деякі аспекти системи, в якій я працював, і поділлюся своїми висновками. Розповім про експерименти, які я проводив, і їхні результати. Давайте надамо трошки контексту. Загалом, це велика система з багатьма базами даних і сервісами, які взаємодіють через різні черги та інші механізми. Однак ми сконцентруємося на модулі, що відповідає за інформацію. Цей модуль інтегрується у загальну систему та відповідає за відправку кампаній через різні канали, такі як електронні листи, текстові повідомлення, мультимедійні повідомлення, пуш-сповіщення та інші. Модуль виконує валідацію, шаблонізацію, обробку відмов, відправку та персоналізацію цих повідомлень.

Цей модуль може бути розгорнутий для кожного клієнта окремо, налаштований дуже тонко і масштабуватися дуже ефективно. Загальна архітектура підсистеми виглядає приблизно так: реквест або заявка надходить, через Kafka, до центральної системи, де всі сервіси взаємодіють через Kafka. Деякі сервіси можуть мати спільну базу даних з цим монолітом, інші можуть мати власні бази даних з різними цілями. Використовується складна схема неймінгу топіків через різні типи компаній та клієнтів.

Тепер трошки про деплоймент. Для цього використовуються Jenkins Groovy та Argo CD. Argo CD дозволяє докладно налаштовувати кожне середовище та керувати нашим Kubernetes кластером через Git. Якщо потрібно внести зміни, ми комітимо їх у Git, Argo виявляє зміни та змінює стан кластера. Тепер перейдемо до сервісу. Цей сервіс відповідає за обробку та відправку повідомлень. Концептуально, він складається з топіку в Kafka з різними партіціями. В середині сервісу є два види завдань: Consumer Task та Worker Task, які взаємодіють через канали. Це дозволяє масштабувати кількість консюмерів та воркерів.

Worker Task відправляє повідомлення на зовнішні провайдери та контролює цей процес. Система використовує Bounded Channel для локального PubSub, щоб управляти потоком повідомлень. Redis використовується для збереження оффсетів, а деякі аспекти архітектури дозволяють ефективно працювати з масштабуванням та змінами в конфігурації. Надіюсь, це вам допоможе. Якщо у вас є ще питання або щось, що ви хочете уточнити, будь ласка, дайте мені знати.

Отже, якщо виникла помилка, ми звертаємося до Redis і виконуємо Consumer Seek, тобто змінюємо offset. Потім ми дивимося на інший Consumer та визначаємо, куди записати ці дані. У нас є масив каналів, кожен з яких пов'язаний з WorkerTask. Ми намагаємося записати в ці канали, але, оскільки bounded канали мають обмежену ємність, ми використовуємо конструкцію Wait to Read і Try Read, щоб уникнути exception, якщо канал вже заповнений. WorkerTask витягує дані з каналу, обробляє їх, персоналізує та вибирає Channel Manager для додаткової обробки.

Але у мережі можуть виникнути лаги, і якщо Worker зупиняється, Konsyumer може накопичити багато повідомлень. Якщо всі Worker зайняті, може виникнути ситуація, коли Konsyumer не можуть розподілити повідомлення. Для вирішення цього ми встановлюємо тайм-аут на команду, і якщо всі Worker зайняті, ми ставимо паузу на партішені, чекаємо, продовжуємо читання та відкатуємося до попереднього повідомлення. Offset зберігається в Redis, група Konsyumer та топік partition включаються.

Механізм партішенів використовується для підвищення паралелізму, а основний дизайн рішення покликаний полегшити обробку повідомлень. Висновки з даного підходу полягають у тому, що Offset зберігаються в окремому сховищі, а зупинки партішенів застосовуються для підстраховки від втрати Konsyumer у випадку навантаження. Bounded-канали дозволяють контролювати доступність та ефективно обмінюватися повідомленнями. Також, Kafka дозволяє обробляти одні й ті ж повідомлення кілька разів, дозволяючи різним сервісам виконувати різні завдання. Цей підхід дозволяє більш гнучко масштабувати та асинхронно будувати систему.

Максимальна гнучкість дозволяє нам створювати різні налаштування для кожної задачі чи каналу, або навіть для кожного типу компанії. Ми можемо контролювати розмір каналу, кількість Worker, інші параметри. Для нових сервісів ми використовуємо minimal API. Замість класичного підходу з контролерами, де кожен контролер відповідав за певний вид view, minimal API надає більшу гнучкість. Ми можемо збирати різні API в різних файлах, використовувати extension методи та інше. З minimal API, ми самі вибираємо, як декомпозувати відповідальності.

Однак, є обмеження, зокрема при роботі з файлами. Використання form файлу всередині моделі може призвести до проблем. Зокрема, при спробі завантажити файл можуть виникнути помилки через несумісність форматів або не саппортовані атрибути. Одним із способів подолати ці проблеми є використання реквеста та контексту для інжектування додаткової метадати в методи. Це дозволяє зручно працювати з MML API, якщо немає контролерів для наслідування. Отриману метадату можна використовувати для зручного отримання значень форми та інших параметрів.

Важливо враховувати обмеження minimal API та шукати кращі способи роботи з певними випадками, такими як завантаження файлів. Перевірка наявності певних атрибутів та коректне використання реквеста та контексту може вирішити деякі проблеми. Отже, ось код для цього Operation Filter. Я видалив зайве. А якщо ми глянемо сюди, то що виходить? Коли ми переглядаємо всі специфікації, ми підходимо до цього endpoint. Та перевіряємо, чи є в його метаданих цей об'єкт. Ця строка - вона є. Добре. Ми переходимо до схеми та просто додаємо ще одне поле, називаємо його Metadata та встановлюємо TypeString. У Swagger це поле з'явиться. Добре, але це виглядає трошки не дуже елегантно, можливо, циклово. Давайте подивимося, чи можна зробити більш універсальне рішення.

Насправді, можна. Давайте створимо складніший об'єкт, який буде зберігати тип та поле, яке потрібно додати. Додамо новий OperationFilter та нову модель, яка буде містити CustomBinding всередині. Це виносить відповідальність в саму модель з тілом метода. Ось трошки коду. Тут вже складніше. Ось це CustomBinding в самій моделі. Тобто ми робимо те саме, але в потрібному місці. А от саме цей фільтр, який ми підключаємо до Swagger. Він отримує всі атрибути з методу за допомогою його Fluent API, який ми можемо додавати в кінці. Отримує атрибути з аргументів методу. Потім перевіряє, чи це атрибути потрібного типу для нас. Тоді ми просто додаємо ще один параметр до цього методу та встановлюємо його тип, який ми задали в конструкторі того атрибута. Просто.

Добре, про це ми поговорили. Давайте перейдемо до групи gRPC та HTTP. У мене були завдання щодо передачі великих файлів. Я не міг використовувати S3 з різних причин, тому мені потрібно було робити це через якийсь endpoint. Та які висновки я зробив. Якщо при конфігуруванні gRPC має більше можливостей з коробки, оскільки це вже річ, яка побудована над HTTP. Наприклад, ось конфігурація gRPC клієнта, де ви можете встановити retry в політиці, включити експоненційний retry, вказати статус-коди для retry, встановити максимальну кількість, налаштувати транспортний рівень тощо.

Далі ми створюємо канал та працюємо з ним. Ось приклад коду. Це Proto3 специфікація для gRPC. Єдиний метод відправляє байти та метадані. Важливо зазначити, що gRPC нічого не знає про файли, він знає тільки про байти. Можна передати файли як байти, але gRPC не контролює, чи був файл переданий повністю та інші аспекти запису файлу. Вам потрібно робити це самостійно, як показано в цьому урізаному коді для демонстрації. На сервері немає єдиного місця, де ми могли б відразу обробити весь файл. Це асинхронний стрім, де з'являється повідомлення при відправці.

Отже, ми перевіряємо, чи прийшла нам метадані. Якщо так, ми виконуємо певні дії, можливо, скануємо команду. Потім перевіряємо, чи прийшла нам data, тобто чи отримали ми які-небудь байти. Якщо так, то ми записуємо ці байти у файловий стрім. Потім потрібно розробити логіку для визначення ваги файлу та інші логічні операції, наприклад, якщо ми передали байти, але не передали метадані. Це вимагає більш складного підходу, оскільки ми можемо передавати будь-що, а gRPC є потужним інструментом для цього. gRPC побудований на протоколі HTTP2, що надає йому більше можливостей. У нього є вбудована система повторних спроб для ситуацій, де можуть виникнути розриви чи артефакти в мережі. Також gRPC може працювати в режимі бідірекційного стріму, де клієнт та сервер можуть взаємодіяти в обидві сторони, що побудовано на протоколі HTTP2.

Крім того, gRPC підтримує протоконтракти, що робить його зручним для використання. Недавно в Postman з'явилася підтримка gRPC, хоча це ще бета-версія. Але вона добре працює. Однак у неї є деякі особливості, такі як кодування файлів у base64 для передачі через Postman, та вона не підтримує API-менеджмент та інші можливості. Щодо HTTP, це широко відомий стандарт з мінімальним порогом входу. Однак HTTP клієнт в .NET не підтримує duplex, хоча сам протокол це дозволяє. Це пов'язано з тим, що HTTP-клієнт призначений для відправлення даних та отримання відповіді, але не для двостороннього зв'язку.

Мені здається, що коли люди намагаються використовувати gRPC при побудові мікросервісних рішень, вони аргументують це тим, що це швидше. Проте, мені здається, це не завжди так. Це може ускладнювати систему, особливо якщо всі компоненти написані на різних технологіях. Іноді можна обійтися просто HTTP, особливо якщо синхронна комунікація необхідна. Але якщо вам дуже важливий високий рівень продуктивності, то, можливо, варто розглянути реалізацію Protobuf для HTTP. У такому випадку можна перейти до синхронної комунікації через стрімінг або меседжинг, якщо необхідно.

Щодо об'єднання різних технологій в одному рішенні, то використання контрактів через sharing може бути дуже зручною рішенням. Бібліотеки в різних технологіях можуть автоматично генерувати код з загальних контрактів, що спрощує роботу з різними технологіями. Моя думка полягає в тому, що коли вам потрібно об'єднати п'ять, десять, або більше технологій в одному рішенні, sharing контрактів є дуже зручним підходом. Це дозволяє легко організувати стрімінг та інші важливі аспекти. Щодо вибору між gRPC та HTTP, обидва вони можуть бути варіантами, залежно від конкретних потреб та важливості продуктивності. Обидва підходи мають свої переваги та недоліки, і вирішення може залежати від конкретного випадку використання. У мене. Все.