Клієнт-серверна архітектура

Асинхронний обмін повідомленнями та протоколи черг

Парадигма асинхронного обміну повідомленнями. Протокол AMQP 0-9-1 та брокер RabbitMQ. Producers, Exchanges, Queues, Bindings. Протокол MQTT для IoT та мобільних пристроїв. Рівні якості обслуговування QoS, топіки та клієнтська інтеграція у Go.

Асинхронний обмін повідомленнями та протоколи черг

🎯 Мета лекції

Зрозуміти фундаментальну різницю між синхронним та асинхронним обміном повідомленнями у розподілених системах.
Опанувати концепцію брокерів повідомлень (message brokers) та їхню роль у відокремленні компонентів системи.
Вивчити протокол AMQP 0-9-1 та архітектурні компоненти RabbitMQ: Producers, Exchanges, Queues, Bindings.
Навчитися проєктувати різні типи обмінників (Direct, Fanout, Topic, Headers) для реалізації патернів комунікації.
Опанувати протокол MQTT як легковаговий механізм обміну повідомленнями для IoT та мобільних пристроїв.
Застосовувати рівні якості обслуговування QoS 0, 1, 2 та механізми гарантованої доставки.

🔑 Ключові терміни

Message Broker: посередник, що приймає повідомлення від виробників та доставляє їх споживачам, забезпечуючи відокремлення (decoupling) компонентів.
AMQP (Advanced Message Queuing Protocol): відкритий стандарт протоколу асинхронного обміну повідомленнями на прикладному рівні.
Producer: компонент, що генерує та надсилає повідомлення до брокера.
Consumer: компонент, що отримує та обробляє повідомлення з черги.
Queue: упорядкована структура даних, що зберігає повідомлення до моменту їх споживання.
Exchange: маршрутизатор повідомлень, що приймає повідомлення від виробників та розподіляє їх у черги згідно з правилами (bindings).
MQTT (Message Queuing Telemetry Transport): легковаговий протокол публікації/підписки, оптимізований для обмежених мереж та пристроїв.

Парадигма асинхронного обміну повідомленнями

Обмеження синхронної комунікації у розподілених системах

На попередніх лекціях ми детально розглянули синхронні протоколи взаємодії: TCP-сокети, HTTP REST API та gRPC. Усі ці підходи реалізують модель запит-відповідь (request-response), де клієнт надсилає запит і блокується (blocks) у очікуванні відповіді від сервера. Ця модель є природною та інтуїтивно зрозумілою — вона відображає спосіб, яким люди спілкуються в повсякденному житті: ми ставимо запитання і чекаємо на відповідь.

Проте у великих розподілених системах, що складаються з десятків або сотень мікросервісів, синхронна комунікація виявляє кілька критичних обмежень, які можуть призвести до каскадних відмов, деградації продуктивності та складності управління життєвим циклом запитів.

Проблема 1: Сильна часова зв'язаність (tight temporal coupling).

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

Розглянемо реальний сценарій інтернет-магазину. Коли користувач розміщує замовлення, система повинна виконати кілька операцій:

Перевірити наявність товару на складі (виклик inventory-service).
Зарезервувати товар та зменшити кількість (виклик inventory-service.Reserve).
Списати кошти з рахунку користувача (виклик payment-service).
Створити запис про замовлення у базі даних (виклик order-service).
Надіслати повідомлення на електронну пошту (виклик email-service).

Якщо всі ці операції виконуються синхронно у ланцюжку, загальний час відповіді становитиме суму часу виконання кожної операції плюс мережеві затримки:

Завантаження...

Припустимо, кожен сервіс відповідає за 50 мілісекунд, а мережева затримка між сервісами становить 5 мілісекунд. Тоді загальний час очікування для користувача складе:

Завантаження...

Це вже сприймається користувачем як повільна відповідь. Але що станеться, якщо email-service тимчасово перевантажений і відповідає за 3 секунди замість 50 мілісекунд? Увесь запит користувача блокується на ці 3 секунди, хоча відправлення електронного листа не є критичною операцією, що повинна блокувати підтвердження замовлення.

Проблема 2: Каскадні відмови та поширення навантаження (cascading failures).

У синхронній моделі повільний або недоступний нижній сервіс може спричинити ефект доміно (domino effect), що поширюється вгору по ланцюжку викликів. Розглянемо ситуацію, коли payment-service раптово починає відповідати повільно (наприклад, через аварію у базі даних). Усі запити до order-service, що чекають на відповідь від payment-service, починають накопичуватися.

Якщо order-service реалізований на Go з моделлю «горутина на запит», кожен заблокований запит утримує горутину та мережеве з'єднання. При навантаженні у 1000 запитів на секунду через 10 секунд у пам'яті буде утримуватися 10 000 заблокованих горутин. Хоча горутини є легковаговими (кілька кілобайтів стека), 10 000 горутин споживатимуть близько 40–80 МБ пам'яті лише на стеки. Додайте до цього буфери TCP-з'єднань, контекстні об'єкти та структури даних запитів — і сервіс може вичерпати пам'ять або дескриптори файлів операційної системи (file descriptor exhaustion).

Коли order-service стає недоступним, фронтенд-додаток, що викликає order-service, також починає накопичувати блоковані запити. Це поширюється на всі вищі рівні системи, поки весь стек не стане нестабільним. Один повільний сервіс може зупинити всю систему.

Для захисту від каскадних відмов у синхронних системах застосовуються патерни стабільності (stability patterns): таймаути (timeouts), обмеження повторних спроб (retry budgets), автомат-вимикач (circuit breaker) та обмеження одночасних запитів (bulkhead isolation). Проте ці механізми лише пом'якшують проблему, але не усувають її корінну причину — сильну зв'язаність у часі.

Проблема 3: Складність керування транзакціями та компенсаційними діями.

Коли операція складається з кількох синхронних викликів до різних сервісів, керування консистентністю даних стає надзвичайно складним завданням. Якщо перші три виклики успішні, а четвертий завершується помилкою, система повинна відкотити зміни у попередніх сервісах.

Наприклад, якщо inventory-service зарезервував товар, payment-service списав кошти, але email-service повернув помилку (наприклад, через недійсну електронну адресу), система повинна:

Викликати payment-service.Refund() для повернення коштів.
Викликати inventory-service.ReleaseReservation() для звільнення товару.

Цей підхід називається патерном Saga із компенсаційними транзакціями (compensating transactions). Реалізація Saga у синхронному коді вимагає складної логіки обробки помилок, збереження стану частково завершених операцій та ризику часткової відмови (що станеться, якщо виклик Refund() також завершиться помилкою?).

Проблема 4: Неможливість обробки сплесків навантаження (traffic spikes).

Синхронна система працює у режимі реального часу: кожен запит повинен бути оброблений негайно. Якщо система отримує раптовий сплеск навантаження (наприклад, під час розпродажу чи вірусного соціального поста), сервер може бути перевантажений запитами, що перевищують його обчислювальні можливості.

У типовій архітектурі сервіс налаштований на певну пропускну здатність (наприклад, 1000 запитів на секунду). Якщо надходить 5000 запитів на секунду, сервер може:

Відхилити запити із статус-кодом 503 Service Unavailable — користувачі бачать помилки.
Поставити запити у внутрішню чергу — але якщо сплеск триває довго, черга переповнюється і пам'ять вичерпується.
Масштабуватися горизонтально (horizontal scaling) — але автоматичне розгортання нових екземплярів сервісу займає 30–120 секунд, протягом яких система перевантажена.

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

Концепція асинхронного обміну повідомленнями

Асинхронний обмін повідомленнями (asynchronous messaging) — це парадигма комунікації, де відправник (sender / producer) надсилає повідомлення до посередника (message broker) і негайно продовжує роботу, не чекаючи на відповідь. Отримувач (receiver / consumer) витягує повідомлення з брокера у зручний для нього момент та обробляє його незалежно від відправника.

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

Loading diagram...

Розглянемо переваги, які надає асинхронна модель:

Перевага 1: Згладжування сплесків навантаження (load leveling).

Брокер повідомлень діє як буфер між виробниками та споживачами. Якщо система отримує 5000 запитів на секунду, але споживач здатен обробити лише 1000 запитів на секунду, повідомлення накопичуються у черзі. Споживач обробляє їх із постійною швидкістю, не перевантажуючись. Коли сплеск навантаження закінчується, черга поступово спорожнюється.

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

Перевага 2: Відмовостійкість через гарантовану доставку (guaranteed delivery).

Сучасні брокери повідомлень (RabbitMQ, Apache Kafka, AWS SQS) забезпечують персистентність повідомлень (message persistence): повідомлення записуються на диск перед підтвердженням прийому. Це означає, що навіть якщо брокер раптово вимкнеться (аварія, перезавантаження), повідомлення не будуть втрачені — вони відновлюються після перезапуску.

Споживач підтверджує успішну обробку повідомлення через механізм підтвердження (acknowledgment / ACK). Якщо споживач завершує роботу аварійно до надсилання ACK, брокер автоматично повертає повідомлення у чергу для повторної обробки іншим споживачем. Це забезпечує гарантію доставки (at-least-once delivery): кожне повідомлення буде оброблено принаймні один раз, навіть у разі відмов.

Перевага 3: Масштабування споживачів незалежно від виробників.

У асинхронній моделі виробники та споживачі є повністю відокремленими (loosely coupled). Виробник не знає, скільки споживачів обробляє повідомлення, і споживач не знає, який виробник надіслав повідомлення. Це дозволяє:

Запускати кілька екземплярів споживача, які конкурують за повідомлення з однієї черги (competing consumers pattern). Брокер автоматично розподіляє повідомлення між екземплярами.
Додавати нові типи споживачів без зміни виробників. Наприклад, якщо потрібно додати аудит усіх створених замовлень, можна запустити новий споживач audit-service, який підписується на ті ж повідомлення.

Перевага 4: Природна реалізація патерну Event-Driven Architecture.

Асинхронний обмін повідомленнями є фундаментом подійно-орієнтованої архітектури (Event-Driven Architecture / EDA). Замість того, щоб один сервіс безпосередньо викликав інші сервіси, він публікує подію (event) про те, що щось сталося (наприклад, OrderCreated, PaymentProcessed, UserRegistered).

Усі зацікавлені сервіси підписуються на ці події та реагують на них незалежно один від одного. Це забезпечує слабку зв'язаність (loose coupling): сервіс order-service не повинен знати, що існують сервіси email-service, analytics-service, warehouse-service, які реагують на подію створення замовлення. Кожен новий сервіс може бути доданий або видалений без змін у order-service.

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

Роль брокерів повідомлень у розподілених системах

Брокер повідомлень (message broker) — це проміжний компонент, що приймає повідомлення від виробників (producers) та доставляє їх споживачам (consumers) згідно з налаштованими правилами маршрутизації. Брокер виконує кілька критичних функцій:

Функція 1: Буферизація та черги.

Брокер зберігає повідомлення у чергах (queues) — упорядкованих структурах даних, що реалізують дисципліну FIFO (First In, First Out). Повідомлення, що надійшло першим, буде оброблено першим. Черги можуть бути персистентними (зберігаються на диску) або тимчасовими (зберігаються у пам'яті).

Функція 2: Маршрутизація повідомлень.

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

Функція 3: Гарантії доставки.

Брокер забезпечує різні рівні гарантій доставки:

At-most-once (щонайбільше один раз): повідомлення може бути втрачене у разі збою. Найшвидший режим, але без гарантій.
At-least-once (принаймні один раз): повідомлення гарантовано буде доставлено, але може бути доставлено кілька разів у разі повторних спроб. Споживач повинен бути ідемпотентним (idempotent) — обробка одного повідомлення кілька разів не повинна призводити до некоректного стану.
Exactly-once (рівно один раз): повідомлення буде доставлено рівно один раз. Найскладніший режим, вимагає розподілених транзакцій.

Функція 4: Відмовостійкість та реплікація.

Сучасні брокери підтримують кластеризацію (clustering) та реплікацію (replication). Повідомлення автоматично реплікуються на кілька вузлів кластера. Якщо один вузол виходить з ладу, інші вузли продовжують обслуговувати запити без втрати даних.

Loading diagram...

Зверніть увагу: кожен споживач працює незалежно. Якщо email-service тимчасово недоступний, повідомлення залишаються у черзі orders.created та чекають на обробку. При цьому warehouse-service продовжує успішно обробляти ті ж повідомлення без затримок. Це забезпечує ізоляцію відмов (failure isolation): проблема одного споживача не впливає на інших.

Протокол AMQP та брокер RabbitMQ

Специфікація протоколу AMQP 0-9-1

AMQP (Advanced Message Queuing Protocol) — це відкритий стандарт протоколу прикладного рівня для асинхронного обміну повідомленнями. Протокол був розроблений консорціумом фінансових установ (JPMorgan Chase, Cisco, Red Hat) у 2006–2008 роках як альтернатива пропрієтарним комерційним брокерам (IBM MQ, TIBCO).

Існує кілька версій специфікації AMQP: AMQP 0-9-1 та AMQP 1.0. Це два принципово різні протоколи, незважаючи на схожість назв. RabbitMQ реалізує версію AMQP 0-9-1, яка є більш розповсюдженою у середовищі мікросервісів. AMQP 1.0 використовується переважно у корпоративних інтеграційних шинах (наприклад, Azure Service Bus, Apache Qpid).

Основні характеристики AMQP 0-9-1:

1. Бінарний протокол із багатошаровою структурою.

AMQP є бінарним протоколом, що працює поверх TCP. Повідомлення AMQP складаються з кількох частин:

Фрейми транспортного рівня (frames): заголовок фрейму містить тип (метод, заголовок, тіло), номер каналу та розмір корисного навантаження.
Методи (methods): команди протоколу (наприклад, basic.publish, basic.consume, queue.declare).
Заголовки повідомлення (message headers): метадані (пріоритет, час життя, тип вмісту).
Тіло повідомлення (message body): корисне навантаження (payload), що може містити JSON, XML, Protocol Buffers або бінарні дані.

2. Віртуальні з'єднання через канали (channels).

AMQP дозволяє мультиплексувати кілька логічних каналів у рамках одного TCP-з'єднання. Кожен канал має власний ідентифікатор та може виконувати незалежні операції (публікація, споживання, оголошення черг). Це зменшує накладні витрати на встановлення з'єднань та дескриптори файлів.

3. Гарантії персистентності та транзакції.

AMQP підтримує підтвердження доставки (acknowledgments) на трьох рівнях:

Publisher Confirms: виробник отримує підтвердження від брокера, що повідомлення було прийняте та збережене.
Consumer Acknowledgments: споживач підтверджує успішну обробку повідомлення.
Транзакції: кілька операцій можуть бути згруповані у транзакцію, що або повністю підтверджується, або повністю скасовується.

4. Модель маршрутизації через обмінники (exchanges).

Ключова інновація AMQP полягає у розділенні логіки публікації та маршрутизації. Виробники не надсилають повідомлення безпосередньо у черги — вони публікують їх до обмінників (exchanges). Обмінник приймає повідомлення та, згідно з налаштованими зв'язками (bindings), розподіляє їх у одну або кілька черг.

Це дозволяє реалізувати складні патерни маршрутизації без зміни коду виробників: всі правила маршрутизації налаштовуються декларативно у конфігурації брокера.

Архітектурні компоненти RabbitMQ

RabbitMQ — найпопулярніша реалізація брокера повідомлень, що підтримує протокол AMQP 0-9-1. RabbitMQ написаний мовою Erlang/OTP — платформою, спеціально розробленою для побудови високодоступних телекомунікаційних систем. Erlang забезпечує легку модель конкурентності (процеси-актори), вбудовану підтримку кластеризації та механізми самовідновлення після збоїв (hot code swapping, supervision trees).

RabbitMQ складається з кількох ключових компонентів, які утворюють архітектурну модель AMQP:

Loading diagram...

Компонент 1: Producer (Виробник).

Виробник — це застосунок, що створює та надсилає повідомлення до RabbitMQ. Виробник встановлює TCP-з'єднання з RabbitMQ, відкриває канал (channel) та публікує повідомлення до обмінника, вказуючи ключ маршрутизації (routing key).

Приклад коду виробника у Go з використанням офіційної бібліотеки amqp091-go:

package main

import (
    "context"
    "log"
    "time"

    amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
    // Встановлення з'єднання з RabbitMQ
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Не вдалося підключитися до RabbitMQ: %v", err)
    }
    defer conn.Close()

    // Відкриття каналу
    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Не вдалося відкрити канал: %v", err)
    }
    defer ch.Close()

    // Оголошення обмінника типу "direct"
    err = ch.ExchangeDeclare(
        "orders",  // назва обмінника
        "direct",  // тип обмінника
        true,      // durable (персистентний)
        false,     // auto-deleted
        false,     // internal
        false,     // no-wait
        nil,       // arguments
    )
    if err != nil {
        log.Fatalf("Не вдалося оголосити обмінник: %v", err)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    // Публікація повідомлення
    body := `{"order_id": "12345", "user_id": "alice", "total": 99.99}`
    err = ch.PublishWithContext(
        ctx,
        "orders",         // exchange
        "order.created",  // routing key
        false,            // mandatory (обов'язкова доставка)
        false,            // immediate
        amqp.Publishing{
            ContentType: "application/json",
            Body:        []byte(body),
            DeliveryMode: amqp.Persistent, // персистентне повідомлення
        },
    )
    if err != nil {
        log.Fatalf("Не вдалося опублікувати повідомлення: %v", err)
    }

    log.Printf("Опубліковано повідомлення: %s", body)
}

Розберемо ключові деталі цього коду:

Рядок підключення: amqp://guest:guest@localhost:5672/ містить логін, пароль, адресу та порт RabbitMQ. У продакшн-середовищі використовуються надійні паролі та SSL/TLS-з'єднання (amqps://).
Канал: це логічне з'єднання всередині TCP-сокета. Один процес може відкрити кілька каналів для паралельної публікації.
ExchangeDeclare: оголошує обмінник. Якщо обмінник уже існує з такими ж параметрами, операція завершується успішно. Якщо параметри відрізняються, повертається помилка.
DeliveryMode: amqp.Persistent: повідомлення буде збережене на диск. Навіть якщо RabbitMQ перезавантажиться, повідомлення залишиться у черзі.

Компонент 2: Exchange (Обмінник).

Обмінник — це маршрутизатор повідомлень усередині RabbitMQ. Обмінник приймає повідомлення від виробників та розподіляє їх у черги згідно з правилами зв'язків (bindings). Обмінник сам не зберігає повідомлення — це лише маршрутизаційна логіка.

RabbitMQ підтримує кілька типів обмінників, кожен із яких реалізує різну стратегію маршрутизації:

Тип 1: Direct Exchange (Прямий обмінник).

Повідомлення доставляється у черги, де ключ зв'язку (binding key) точно збігається з ключем маршрутизації (routing key) повідомлення. Це найпростіший та найшвидший тип обмінника.

Сценарій використання: розподіл повідомлень за типом події. Наприклад:

Повідомлення з routing_key="order.created" йде у чергу order-processing-queue.
Повідомлення з routing_key="order.cancelled" йде у чергу refund-queue.

Loading diagram...

Тип 2: Fanout Exchange (Широкомовний обмінник).

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

Сценарій використання: широкомовна розсилка (broadcast). Наприклад, подія user.registered повинна бути оброблена трьома сервісами:

email-service — надішле вітальний лист.
analytics-service — оновить статистику реєстрацій.
crm-service — створить запис про клієнта у CRM-системі.

Loading diagram...

Код оголошення Fanout Exchange та зв'язку черг:

// Оголошення Fanout обмінника
err = ch.ExchangeDeclare(
    "user-events", // назва
    "fanout",      // тип
    true,          // durable
    false,         // auto-deleted
    false,         // internal
    false,         // no-wait
    nil,
)

// Оголошення черги для email-service
qEmail, err := ch.QueueDeclare(
    "email-queue", // назва черги
    true,          // durable
    false,         // auto-delete
    false,         // exclusive
    false,         // no-wait
    nil,
)

// Зв'язування черги з обмінником (routing key ігнорується)
err = ch.QueueBind(
    qEmail.Name,    // назва черги
    "",             // routing key (порожній для fanout)
    "user-events",  // обмінник
    false,
    nil,
)

Тип 3: Topic Exchange (Тематичний обмінник).

Найгнучкіший тип обмінника. Повідомлення маршрутизуються згідно з шаблонами (patterns) у ключі зв'язку. Ключ маршрутизації має структуру із сегментів, розділених крапками (наприклад, region.service.action), і черги можуть підписуватися на шаблони з використанням підстановочних символів:

* (зірочка) — замінює рівно один сегмент.
# (решітка) — замінює нуль або більше сегментів.

Приклади шаблонів:

order.* — відповідає order.created, order.updated, але не order.item.added.
order.# — відповідає order.created, order.item.added, order.item.removed.123.
*.critical — відповідає payment.critical, database.critical.

Сценарій використання: логування та моніторинг. Система генерує логи з різними рівнями важливості та джерелами:

auth.info — інформаційні події автентифікації.
auth.error — помилки автентифікації.
payment.critical — критичні події платіжного модуля.

Можна налаштувати кілька споживачів:

Споживач з шаблоном *.critical отримує лише критичні події з усіх модулів.
Споживач з шаблоном auth.# отримує всі події модуля автентифікації.
Споживач з шаблоном # отримує всі події (повний аудит).

Loading diagram...

Тип 4: Headers Exchange (Обмінник за заголовками).

Маршрутизація відбувається не на основі routing key, а на основі заголовків повідомлення (message headers). Виробник передає набір пар ключ-значення у заголовках, а черга вказує умови відповідності (наприклад, format=json AND priority=high).

Цей тип обмінника використовується рідко, оскільки він складніший та повільніший за інші типи. Його застосовують, коли потрібна маршрутизація за кількома атрибутами одночасно, а структура routing_key стає громіздкою.

Компонент 3: Queue (Черга).

Черга — це впорядкований буфер, що зберігає повідомлення до моменту їх споживання. Черги мають кілька важливих атрибутів:

Durable (Персистентна черга).

Якщо черга оголошена як durable, її метаінформація зберігається на диск. Після перезавантаження RabbitMQ черга відновлюється. Проте це не гарантує збереження самих повідомлень — для цього потрібно публікувати повідомлення з прапором DeliveryMode: amqp.Persistent.

Exclusive (Ексклюзивна черга).

Ексклюзивна черга може бути використана лише одним з'єднанням. Коли з'єднання закривається, черга автоматично видаляється. Використовується для тимчасових черг (наприклад, для RPC-викликів з унікальною чергою відповідей).

Auto-delete (Автоматичне видалення).

Черга автоматично видаляється, коли останній споживач від'єднується. Використовується для тимчасових задач.

TTL (Time To Live, Час життя).

Повідомлення у черзі можуть мати обмежений час життя. Якщо повідомлення не було спожите протягом TTL, воно автоматично видаляється або переміщується у Dead Letter Queue (черга мертвих листів).

Код оголошення черги з параметрами:

q, err := ch.QueueDeclare(
    "order-processing", // назва черги
    true,               // durable
    false,              // auto-delete
    false,              // exclusive
    false,              // no-wait
    amqp.Table{
        "x-message-ttl":          60000, // TTL 60 секунд (мілісекунди)
        "x-max-length":           10000, // максимум 10000 повідомлень
        "x-dead-letter-exchange": "dlx", // Dead Letter Exchange
    },
)

Компонент 4: Consumer (Споживач).

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

Режим 1: Push (сервер надсилає повідомлення споживачу).

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

Приклад коду споживача у Go:

package main

import (
    "log"
    "time"

    amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Не вдалося підключитися: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Не вдалося відкрити канал: %v", err)
    }
    defer ch.Close()

    // Встановлення QoS (prefetch count)
    err = ch.Qos(
        1,     // prefetch count (кількість неопрацьованих повідомлень)
        0,     // prefetch size
        false, // global
    )
    if err != nil {
        log.Fatalf("Не вдалося встановити QoS: %v", err)
    }

    // Підписка на чергу
    msgs, err := ch.Consume(
        "order-processing", // назва черги
        "",                 // consumer tag (автогенерується)
        false,              // auto-ack (вимкнено, підтверджуємо вручну)
        false,              // exclusive
        false,              // no-local
        false,              // no-wait
        nil,
    )
    if err != nil {
        log.Fatalf("Не вдалося підписатися: %v", err)
    }

    log.Println("Споживач запущено. Очікування повідомлень...")

    // Обробка повідомлень у циклі
    for msg := range msgs {
        log.Printf("Отримано повідомлення: %s", string(msg.Body))

        // Симуляція обробки
        time.Sleep(2 * time.Second)
        log.Println("Повідомлення оброблено")

        // Підтвердження успішної обробки
        err := msg.Ack(false)
        if err != nil {
            log.Printf("Помилка підтвердження: %v", err)
        }
    }
}

Розглянемо важливі деталі:

Prefetch Count (QoS).

Параметр Qos(1, 0, false) обмежує кількість неопрацьованих повідомлень, що можуть бути доставлені споживачу одночасно. Це критично для балансування навантаження між кількома споживачами.

Якщо один споживач працює повільніше за інший (наприклад, через навантаження на процесор або мережеві затримки), RabbitMQ автоматично надсилає більше повідомлень швидшому споживачу. Без Prefetch Count RabbitMQ може надіслати всі повідомлення одному споживачу заздалегідь, залишивши інших простоювати.

Manual Acknowledgment (Ручне підтвердження).

Параметр auto-ack=false означає, що споживач повинен вручну підтвердити обробку повідомлення викликом msg.Ack(false). Якщо споживач завершує роботу аварійно до виклику Ack, RabbitMQ автоматично повертає повідомлення у чергу для повторної обробки іншим споживачем.

Це забезпечує гарантію at-least-once delivery: кожне повідомлення буде оброблено принаймні один раз, навіть у разі збоїв.

Споживач повинен бути ідемпотентним (idempotent): обробка одного повідомлення кілька разів не повинна призводити до некоректного стану. Наприклад, якщо повідомлення містить команду «нарахувати 10 балів користувачу», повторна обробка не повинна нараховувати ще 10 балів. Використовуйте унікальні ідентифікатори повідомлень (message ID) та перевіряйте, чи повідомлення вже було оброблено раніше.

Режим 2: Pull (споживач запитує повідомлення вручну).

Споживач викликає метод ch.Get() для отримання одного повідомлення. Це синхронний режим, що використовується рідко — переважно для одноразових задач або налагодження.

Компонент 5: Bindings (Зв'язки).

Зв'язок (binding) — це правило, що визначає, як обмінник доставляє повідомлення у чергу. Зв'язок складається з:

Назви обмінника.
Назви черги.
Ключа зв'язку (binding key) — шаблон для Direct або Topic обмінників.

Одна черга може бути зв'язана з кількома обмінниками, і один обмінник може маршрутизувати у кілька черг. Це дозволяє будувати складні топології маршрутизації.

Патерни використання RabbitMQ

Патерн 1: Work Queue (Черга завдань).

Кілька споживачів конкурують за повідомлення з однієї черги. Кожне повідомлення обробляється лише одним споживачем. Використовується для розподіленої обробки фонових завдань.

Приклад: система генерації звітів. Користувачі запитують звіти, які додаються у чергу report-generation-queue. Кілька воркерів (worker-1, worker-2, worker-3) паралельно обробляють звіти, вибираючи їх з черги.

Патерн 2: Publish/Subscribe (Публікація/Підписка).

Одне повідомлення доставляється кільком споживачам через Fanout Exchange. Кожен споживач отримує копію повідомлення у свою чергу.

Приклад: подія order.created обробляється трьома сервісами незалежно.

Патерн 3: Routing (Маршрутизація).

Повідомлення доставляється у черги згідно з ключем маршрутизації через Direct Exchange.

Приклад: система обробки логів, де логи різних рівнів (info, warning, error) розподіляються у різні черги для окремої обробки.

Патерн 4: Topics (Тематична маршрутизація).

Повідомлення маршрутизуються згідно з шаблонами через Topic Exchange.

Приклад: моніторинг мікросервісів. Події мають структуру service.level (наприклад, auth.critical, payment.info). Споживачі підписуються на шаблони *.critical для отримання критичних подій з усіх сервісів.

Патерн 5: RPC (Remote Procedure Call).

RabbitMQ може бути використаний для реалізації RPC-викликів. Клієнт публікує повідомлення-запит у чергу запитів, вказуючи у заголовку reply_to назву тимчасової черги для відповіді. Сервер обробляє запит та публікує результат у чергу відповідей.

Цей підхід використовується рідко, оскільки gRPC є більш ефективним для синхронних RPC-викликів.

RabbitMQ підтримує Dead Letter Exchange (DLX) — обмінник, куди автоматично переміщуються повідомлення, що не вдалося обробити (наприклад, через вичерпання TTL, відхилення споживачем або досягнення максимальної кількості повторних спроб). DLX використовується для аналізу проблемних повідомлень та налагодження.

Протокол MQTT для IoT та мобільних пристроїв

Специфікація протоколу MQTT та його призначення

MQTT (Message Queuing Telemetry Transport) — це легковаговий протокол обміну повідомленнями на основі моделі публікація/підписка (publish/subscribe), розроблений спеціально для обмежених мереж із низькою пропускною здатністю, високими затримками та ненадійними з'єднаннями.

Протокол MQTT був розроблений у 1999 році компаніями IBM та Arcom (нині Eurotech) для моніторингу нафтопроводів через супутниковий зв'язок. Пізніше протокол став стандартом OASIS (2013) та ISO/IEC (2016). Сьогодні MQTT є основним протоколом для IoT-пристроїв (Internet of Things): розумні будинки, промислові сенсори, автомобільна телеметрія, медичні пристрої.

Основні характеристики MQTT:

1. Мінімальний розмір повідомлення.

Заголовок MQTT-повідомлення займає лише 2 байти у мінімальній конфігурації. Для порівняння, мінімальний HTTP-заголовок займає близько 200–300 байтів. Це критично для IoT-пристроїв, що працюють у мережах із обмеженою пропускною здатністю (2G/3G мобільні мережі, LoRaWAN, Zigbee).

2. Модель публікація/підписка з топіками.

MQTT не має концепції черг (queues), як у AMQP. Замість цього, повідомлення публікуються у топіки (topics) — ієрархічні рядки, розділені слешами (наприклад, home/livingroom/temperature, factory/machine5/vibration). Клієнти підписуються на топіки за допомогою точних імен або підстановочних символів (+ та #).

3. Три рівні якості обслуговування (QoS).

MQTT підтримує три рівні гарантій доставки, що дозволяють балансувати між надійністю та продуктивністю:

QoS 0 (At most once): Повідомлення надсилається один раз без підтвердження. Може бути втрачене. Найшвидший режим.
QoS 1 (At least once): Повідомлення доставляється принаймні один раз. Отримувач підтверджує доставку пакетом PUBACK. Може бути дубльоване.
QoS 2 (Exactly once): Повідомлення доставляється рівно один раз через чотирирівневе рукостискання (PUBLISH, PUBREC, PUBREL, PUBCOMP). Найповільніший, але найнадійніший режим.

4. Легковаговий протокол Keep-Alive.

MQTT підтримує механізм Keep-Alive: клієнт та брокер періодично обмінюються пакетами PINGREQ/PINGRESP, щоб перевірити живість з'єднання. Це дозволяє швидко виявляти обірвані з'єднання та економити заряд батареї (клієнт не витрачає енергію на тривале очікування таймаута TCP).

5. Last Will and Testament (LWT) — Заповіт.

Клієнт може зареєструвати заповідальне повідомлення (Last Will), яке брокер автоматично публікує у вказаний топік, якщо клієнт раптово від'єднався (наприклад, через мережевий збій або розрядження батареї). Це дозволяє іншим клієнтам дізнатися про втрату зв'язку з пристроєм.

Loading diagram...

Структура топіків та підстановочні символи

Топіки у MQTT є ієрархічними рядками, де рівні ієрархії розділяються символом / (слеш). Це схоже на структуру файлової системи або URL-шляхів.

Приклади топіків:

home/kitchen/temperature
factory/floor2/machine15/vibration
vehicles/truck123/gps/latitude

Клієнти можуть підписуватися на топіки за допомогою підстановочних символів (wildcards):

Підстановочний символ + (плюс) — один рівень.

Замінює рівно один рівень ієрархії.

Підписка home/+/temperature відповідає:
- home/kitchen/temperature ✅
- home/bedroom/temperature ✅
- home/kitchen/humidity ❌ (інше поле)
- home/kitchen/sensors/temperature ❌ (два рівні замість одного)

Підстановочний символ # (решітка) — багато рівнів.

Замінює нуль або більше рівнів ієрархії. Повинен бути останнім символом у підписці.

Підписка home/# відповідає:
- home/kitchen/temperature ✅
- home/bedroom/humidity ✅
- home/kitchen/sensors/air_quality ✅
- home ✅ (нуль рівнів після home)
- office/kitchen/temperature ❌ (інший корінь)
Підписка factory/floor2/# відповідає всім повідомленням з другого поверху фабрики.

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

Рівні якості обслуговування (QoS)

Вибір рівня QoS є критичним компромісом між надійністю, продуктивністю та витратами мережевого трафіку.

QoS 0: At most once (Щонайбільше один раз).

Повідомлення надсилається один раз без підтвердження. Це аналог UDP у транспортному рівні — швидко, але без гарантій.

Loading diagram...

Використовується для:

Телеметрія сенсорів у реальному часі, де втрата окремого повідомлення не критична (температура, вологість).
Високочастотні оновлення (GPS-координати кожної секунди).

QoS 1: At least once (Принаймні один раз).

Повідомлення доставляється з підтвердженням. Якщо підтвердження не отримано протягом таймаута, повідомлення надсилається повторно. Може призвести до дублювання.

Loading diagram...

Використовується для:

Команди керування IoT-пристроями (увімкнути світло, змінити температуру).
Події, що вимагають обов'язкової обробки, але дублювання допустиме (оновлення статусу).

QoS 2: Exactly once (Рівно один раз).

Найнадійніший режим із чотирирівневим рукостисканням. Гарантує, що повідомлення буде доставлено рівно один раз, без втрат та дублювання.

Loading diagram...

Використовується для:

Фінансові транзакції (списання коштів).
Критичні команди, де дублювання неприпустиме (відкрити електронний замок, запустити двигун).

QoS 2 створює значні накладні витрати: кожне повідомлення вимагає чотирьох мережевих обмінів. Використовуйте його лише для критичних повідомлень, де дублювання або втрата неприпустимі.

Механізми Retained Messages та Clean Session

Retained Messages (Збережені повідомлення).

Коли повідомлення публікується з прапором retained=true, брокер зберігає останнє повідомлення для цього топіка. Усі нові підписники, які підпишуться на топік пізніше, негайно отримують збережене повідомлення.

Приклад: топік home/livingroom/light/state містить стан лампи (ON або OFF). Коли нова панель керування підключається до брокера, вона негайно отримує останній стан лампи, не чекаючи на наступну зміну.

client.Publish(
    "home/livingroom/light/state", // топік
    1,                              // QoS
    true,                           // retained
    "ON",                           // payload
)

Clean Session (Чиста сесія).

Параметр CleanSession визначає, чи брокер повинен зберігати інформацію про підписки клієнта після від'єднання:

CleanSession=true: Брокер видаляє всі підписки та черги повідомлень після від'єднання. Клієнт починає «з чистого аркуша» при кожному підключенні.
CleanSession=false: Брокер зберігає підписки та накопичує повідомлення QoS 1/2 під час відсутності клієнта. Коли клієнт повертається, він отримує всі пропущені повідомлення.

Використовується для мобільних застосунків, які періодично втрачають зв'язок, але повинні отримувати всі сповіщення після відновлення з'єднання.

Реалізація MQTT-клієнта у Go на базі Eclipse Paho

Бібліотека Eclipse Paho є офіційною клієнтською реалізацією MQTT для кількох мов програмування, включно з Go.

Встановлення бібліотеки:

go get github.com/eclipse/paho.mqtt.golang

Приклад MQTT Publisher (Виробник):

package main

import (
    "fmt"
    "log"
    "time"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

func main() {
    // Налаштування параметрів підключення
    opts := mqtt.NewClientOptions()
    opts.AddBroker("tcp://localhost:1883") // Адреса MQTT брокера (Eclipse Mosquitto)
    opts.SetClientID("temperature_sensor_01")
    opts.SetUsername("user")
    opts.SetPassword("password")
    opts.SetKeepAlive(60 * time.Second) // Keep-Alive інтервал
    
    // Налаштування Last Will (заповідальне повідомлення)
    opts.SetWill(
        "sensors/status",        // топік
        "sensor_01 disconnected", // payload
        1,                       // QoS
        false,                   // retained
    )

    // Створення клієнта
    client := mqtt.NewClient(opts)
    
    // Підключення до брокера
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        log.Fatalf("Помилка підключення: %v", token.Error())
    }
    defer client.Disconnect(250)

    log.Println("Підключено до MQTT брокера")

    // Публікація повідомлень у циклі
    for i := 0; i < 10; i++ {
        temperature := 20.0 + float64(i)*0.5
        payload := fmt.Sprintf("%.1f", temperature)

        token := client.Publish(
            "home/livingroom/temperature", // топік
            1,                             // QoS 1
            false,                         // retained
            payload,
        )
        token.Wait() // Очікування підтвердження

        if token.Error() != nil {
            log.Printf("Помилка публікації: %v", token.Error())
        } else {
            log.Printf("Опубліковано: %s°C", payload)
        }

        time.Sleep(2 * time.Second)
    }
}

Приклад MQTT Subscriber (Споживач):

package main

import (
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

// Callback-функція для обробки вхідних повідомлень
var messageHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    log.Printf("Отримано повідомлення з топіка %s: %s", msg.Topic(), string(msg.Payload()))
    
    // Симуляція обробки
    // ... логіка обробки даних ...
}

// Callback-функція при успішному підключенні
var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
    log.Println("Підключено до MQTT брокера")
    
    // Підписка на топік після підключення
    token := client.Subscribe("home/+/temperature", 1, nil)
    token.Wait()
    
    if token.Error() != nil {
        log.Fatalf("Помилка підписки: %v", token.Error())
    }
    log.Println("Підписано на топік: home/+/temperature")
}

// Callback-функція при втраті з'єднання
var connectionLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
    log.Printf("З'єднання втрачено: %v", err)
}

func main() {
    opts := mqtt.NewClientOptions()
    opts.AddBroker("tcp://localhost:1883")
    opts.SetClientID("dashboard_client")
    opts.SetUsername("user")
    opts.SetPassword("password")
    opts.SetCleanSession(false) // Зберігати підписки між сесіями
    
    // Реєстрація callback-функцій
    opts.SetDefaultPublishHandler(messageHandler)
    opts.OnConnect = connectHandler
    opts.OnConnectionLost = connectionLostHandler
    opts.SetAutoReconnect(true) // Автоматичне перепідключення

    client := mqtt.NewClient(opts)
    
    if token := client.Connect(); token.Wait() && token.Error() != nil {
        log.Fatalf("Помилка підключення: %v", token.Error())
    }

    // Очікування сигналу завершення
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
    <-sigChan

    log.Println("Відключення від брокера...")
    client.Disconnect(250)
}

Розглянемо ключові деталі реалізації:

Callback-функції.

MQTT-клієнт є асинхронним. Замість блокуючого очікування повідомлень у циклі, клієнт реєструє callback-функції, які викликаються автоматично при надходженні подій:

OnConnect — викликається після успішного підключення (або перепідключення).
OnConnectionLost — викликається при втраті з'єднання.
DefaultPublishHandler — викликається для кожного отриманого повідомлення.

Автоматичне перепідключення.

Параметр SetAutoReconnect(true) дозволяє клієнту автоматично відновлювати з'єднання у разі мережевих збоїв. Це критично для IoT-пристроїв, що працюють у ненадійних мережах.

Підписка після підключення.

Підписка виконується у callback-функції OnConnect, а не у основному коді. Це гарантує, що підписка буде відновлена після кожного перепідключення (оскільки брокер не зберігає підписки для клієнтів із CleanSession=true).

Порівняння AMQP (RabbitMQ) та MQTT

Критерій	AMQP (RabbitMQ)	MQTT
Призначення	Корпоративні мікросервіси, фонові завдання	IoT-пристрої, мобільні застосунки
Модель комунікації	Черги (Queues) + Обмінники (Exchanges)	Топіки (Topics)
Розмір заголовка	~8 байтів (мінімум)	2 байти (мінімум)
Гарантії доставки	At-least-once, Exactly-once (через транзакції)	QoS 0, 1, 2
Персистентність	Підтримується (черги + повідомлення)	Retained Messages, Clean Session
Складність	Високо (обмінники, зв'язки, політики)	Низька (топіки, підписки)
Використання пам'яті	Високе (складні структури маршрутизації)	Низьке (оптимізовано для обмежених пристроїв)
Протокол транспорту	TCP (бінарний)	TCP (бінарний), WebSocket
Реалізації брокерів	RabbitMQ, Apache Qpid, Azure Service Bus	Eclipse Mosquitto, HiveMQ, AWS IoT Core

У великих системах часто використовуються обидва протоколи одночасно: MQTT для збору телеметрії з IoT-пристроїв, а RabbitMQ для обробки даних на бекенді та координації між мікросервісами. Наприклад, IoT-сенсори публікують дані у MQTT-брокер, а спеціальний міст (bridge) транслює повідомлення з MQTT у RabbitMQ для подальшої обробки аналітичними сервісами.

Практичні сценарії застосування асинхронного обміну

Сценарій 1: Фонова обробка завдань (Background Job Processing)

Проблема: Користувач завантажує відео на платформу. Система повинна виконати кілька операцій: конвертувати відео у кілька форматів, згенерувати мініатюри, витягти метадані, сканувати на заборонений контент.

Рішення: Замість виконання всіх операцій синхронно (що зайняло б 2–5 хвилин), сервер негайно повертає користувачу відповідь «Відео прийнято, обробка розпочнеться незабаром», публікує завдання у RabbitMQ чергу video-processing-queue, та кілька воркерів обробляють завдання паралельно.

Переваги:

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

Сценарій 2: Телеметрія IoT-пристроїв у реальному часі

Проблема: Розумний будинок містить 50 сенсорів (температура, вологість, освітленість, датчики руху). Кожен сенсор генерує оновлення кожні 5 секунд. Це 600 повідомлень на хвилину.

Рішення: Усі сенсори публікують телеметрію у MQTT-брокер. Мобільний застосунок підписується на топіки кімнат, де перебуває користувач, та отримує оновлення у реальному часі. Сервер аналітики підписується на всі топіки (home/#) та зберігає історичні дані.

Переваги:

Мінімальне використання мережевого трафіку (малий розмір повідомлень MQTT).
Сенсори не блокуються у очікуванні відповіді.
Нові споживачі можуть бути додані без змін у сенсорах.

Сценарій 3: Подійно-орієнтована архітектура (Event-Driven Architecture)

Проблема: У системі електронної комерції при створенні замовлення потрібно:

Відправити email користувачу.
Оновити інвентар на складі.
Створити рахунок-фактуру у бухгалтерській системі.
Надіслати сповіщення менеджеру.
Записати подію в аналітику.

Рішення: Сервіс order-service публікує подію OrderCreated у RabbitMQ Fanout Exchange. Кожен зацікавлений сервіс (email-service, inventory-service, billing-service, notification-service, analytics-service) має власну чергу та обробляє подію незалежно.

Переваги:

order-service не знає про існування інших сервісів.
Нові сервіси можуть бути додані без змін у order-service.
Відмова одного сервісу не впливає на інші.

Підсумок та ключові висновки

✅ Основні концепції

Асинхронний обмін повідомленнями відокремлює виробників та споживачів у часі, дозволяючи кожному компоненту працювати незалежно.
Брокери повідомлень (RabbitMQ, MQTT) забезпечують буферизацію, маршрутизацію, персистентність та гарантії доставки.
AMQP є протоколом для корпоративних мікросервісів із складною логікою маршрутизації через обмінники та черги.
MQTT є легковаговим протоколом для IoT-пристроїв та мобільних застосунків із мінімальним використанням мережі.

🎯 Практичне застосування

Використовуйте Work Queues для розподіленої обробки фонових завдань.
Застосовуйте Publish/Subscribe для подійно-орієнтованої архітектури, де кілька сервісів реагують на одну подію.
Використовуйте Topic Exchange для гнучкої маршрутизації повідомлень за шаблонами.
Застосовуйте MQTT для збору телеметрії IoT-пристроїв із мінімальним енергоспоживанням.

⚠️ Пастки та обмеження

Асинхронність ускладнює налагодження: неможливо простежити виконання запиту від початку до кінця.
Споживачі повинні бути ідемпотентними: повторна обробка повідомлення не повинна призводити до некоректного стану.
Черги не замінюють бази даних: використовуйте їх для координації, а не для довготривалого зберігання даних.
QoS 2 у MQTT створює значні накладні витрати — використовуйте лише для критичних команд.

У наступній лекції ми розглянемо протоколи потокового медіа та технологію WebRTC: HLS, DASH, адаптивний бітрейт, P2P-комунікації та реалізація відеозв'язку у Go.

Edit this pageorReport an issue

Міжсервісна взаємодія на основі протоколу gRPC

Концепція віддаленого виклику процедур (RPC). Формат бінарної серіалізації Protocol Buffers. Транспорт HTTP/2 у gRPC. Типи викликів: Unary, Server Streaming, Client Streaming, Bidirectional Streaming. Реалізація сервісів, інтерцептори та генерація клієнтських стабів.

Протоколи потокового медіа та технологія WebRTC

Глибоке вивчення принципів потокової передачі медіаконтенту через HTTP (HLS, DASH), адаптивного бітрейту та технології прямого P2P зв