Цитата з офіційної головної сторінки:

Дві ролі в одному продукті

• Messaging broker - класичні черги повідомлень (AMQP 0-9-1, AMQP 1.0, MQTT, STOMP).
• Streaming broker - append-only логи з offset-based consume (Streams), для event sourcing і replay.

Позиціонування з тієї ж сторінки: "ideal for distributed microservices, real-time data, and IoT".

AMQP = Advanced Message Queuing Protocol.

Wire-level vs API-level

AMQP - це wire-level протокол (бінарний, по TCP), а не API. Wire-level означає, що специфікація фіксує точний формат байтів, які летять по мережі ("over the wire"): фрейми, opcodes (коди операцій), кодування типів.

Контраст - API-level (як JDBC, PDO): описує функції бібліотеки, а як саме воно йде по мережі - справа драйвера.

Дві несумісні версії в одному broker'і

• AMQP 0-9-1 - історично основний у RabbitMQ, на ньому всі PHP-клієнти (php-amqplib, ext-amqp). Курс йде по 0-9-1.
• AMQP 1.0 - окремий протокол OASIS-стандарту (інший дизайн, не "оновлення" 0-9-1). У RabbitMQ 4.0+ став core. У курсі - тільки згадка.

Чотири канонічні сценарії з офіційної головної (rabbitmq.com).

Що дає брокер посередині

• Producer і consumer не мусять бути одночасно онлайн.
• Один producer → багато consumer'ів (і навпаки), без взаємного знання.
• На брокері - persistence (зберігання на диск), retry-логіка, маршрутизація.

Queue - не "просто черга". При queue_declare задається 4 незалежних прапори, які визначають lifecycle і поведінку.

Як вони комбінуються

Окрім прапорів, у 4.x queue має ще тип - спосіб реалізації. Задається при declare через x-queue-type argument. Default історично - classic, але production-стандарт у 4.x - quorum.

Як задавати тип

$channel->queue_declare(
    'orders', false, true, false, false, false,
    new AMQPTable(['x-queue-type' => 'quorum'])  // ← обов'язково для production
);

Базові queue arguments (огляд, deep-dive у пізніших розділах)

Routing key (ключ маршрутизації) - рядок, який producer додає до кожного publish. Адресна "етикетка" повідомлення.

Хто його ставить

• Producer - вказує routing key у виклику basic_publish(body, exchange, routing_key).
• Binding - має свій binding key (інший рядок). Exchange матчить routing key повідомлення проти binding key за правилом, специфічним для типу exchange.

Як трактується різними типами exchange

Unicast (одноадресна доставка) - "одне повідомлення йде в одну конкретну точку". Контраст: broadcast (всім - радіо) - це fanout; multicast (кільком конкретним) - підмножина.

Алгоритм

З документації: "When a new message with routing key R arrives at the direct exchange, the exchange routes it to the queue if K = R", де K - binding key. Якщо в кількох queues однаковий K - всі отримають копію.

Default exchange (нюанс)

Алгоритм: fanout "routes messages to all of the queues that are bound to it and the routing key is ignored". При publish "a copy of the message is delivered to all N queues".

Use cases (з документації)

• MMO-ігри: оновлення лідерборду і глобальні події всім гравцям.
• Спортивні сайти: realtime-оновлення рахунку мобільним клієнтам.
• Розподілені системи: broadcast стану і конфігурацій усім нодам.
• Групові чати: розсилка повідомлень учасникам.

Жорстка вимога до routing key (з tutorial-five-php): "it must be a list of words, delimited by dots". Довжина: до 255 байт.

Wildcards (підстановочні символи у binding key)

• * "can substitute for exactly one word".
• # "can substitute for zero or more words".

Приклади matching

Use cases

• Геолокація: point_of_sale.eu.de.berlin.
• Workers за типом задачі: task.image.resize, task.video.transcode.
• Stocks/News по категоріях. Build-системи: build.linux.x86_64.

Як працює

• Ігнорує routing_key.
• Маршрутизує за message headers (заголовки повідомлення - пари ключ-значення, метадані поверх body).
• "A message matches when the value of the header equals the value specified upon binding".

Аргумент binding'а x-match

• x-match: any - "just one matching header value is sufficient".
• x-match: all - "all the values must match" (default).
• any-with-x / all-with-x - те саме, але враховуються headers що починаються з x-. За замовчуванням x- headers ігноруються.

Connection (з'єднання)

• "AMQP 0-9-1 is an application level protocol that uses TCP for reliable delivery".
• "AMQP 0-9-1 connections are typically long-lived" - відкриваєш один раз при старті worker'а.

Channel (канал)

Multiplexing (мультиплексування) - "кілька логічних потоків живуть в одному фізичному каналі зв'язку, кожен зі своїм id". Один TCP - багато паралельних "логічних з'єднань".

Чому через channels, а не багато connections

TCP - дорогий ресурс (file descriptors, kernel-стан, TLS-handshake, NAT/firewall). Channels - дешеві логічні id, тисячі на одне з'єднання.

services:
  rabbitmq:
    image: rabbitmq:4.3-management
    container_name: rabbitmq-study
    hostname: rabbitmq-study
    ports:
      - "5672:5672"     # AMQP 0-9-1 / 1.0
      - "15672:15672"   # management UI + HTTP API
      - "15692:15692"   # Prometheus metrics
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest
      RABBITMQ_DEFAULT_VHOST: /
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 10s
      timeout: 5s
      retries: 10
      start_period: 30s
    restart: unless-stopped

volumes:
  rabbitmq-data:

• image: rabbitmq:4.3-management - офіц. документація: "second set of tags provided with the management plugin installed and enabled by default". Без -management - broker без UI на 15672.
• hostname: rabbitmq-study - критично. Erlang-нода прив'язана до nodename rabbit@<hostname>. Зміна hostname → broker не знайде свою data в томі.
• volumes: rabbitmq-data:/var/lib/rabbitmq - named volume (постійне сховище Docker, переживає рестарти і recreate). Без volume - docker compose down стирає все.
• healthcheck - Docker сам викликає rabbitmq-diagnostics -q ping кожні 10 с. start_period: 30s - перші 30 секунд невдачі не рахуються (broker довго стартує). Важливо для depends_on: condition: service_healthy у multi-container setup.
• restart: unless-stopped - перезапуск після crash або reboot хоста; не перезапускає, якщо ми зупинили вручну.

Команди

cd /home/serhii/study/courses/rabbitmq/sandbox

# Pull image (~250MB) і запуск у detached mode
docker compose up -d

# Перевірити стан контейнера
docker compose ps

# Лог-стрім (Ctrl+C для виходу, контейнер працює далі)
docker compose logs -f rabbitmq

Очікуваний вивід docker compose ps

NAME             IMAGE                     COMMAND                   STATUS
rabbitmq-study   rabbitmq:4.3-management   "docker-entrypoint.s…"    Up 35s (healthy)

Що відбувається при першому запуску

1. Docker pull rabbitmq:4.3-management з Docker Hub.
2. Створюється named volume rabbitmq-data (порожній).
3. Контейнер стартує, entrypoint застосовує env-змінні (створює user guest, vhost /).
4. Erlang VM піднімає broker, ініціалізує Khepri-store на диску.
5. Healthcheck після 30 с (start_period) починає викликати rabbitmq-diagnostics ping.
6. Коли ping проходить - status стає healthy.

Як зупинити / перезапустити

docker compose stop          # стоп без видалення контейнера
docker compose start         # старт існуючого контейнера
docker compose restart       # alias

docker compose down          # стоп + видалення контейнера (volume лишається!)
docker compose down -v       # ⚠ + видалення volume - стирає УСІ дані

rabbitmq-diagnostics - офіційний CLI broker'а. Команда ping перевіряє, що нода відповідає на Erlang-distribution rpc-виклик.

docker compose exec rabbitmq rabbitmq-diagnostics -q ping
# Pong

Прапор -q (quiet) виводить тільки результат без banner'а. Exit code 0 = ping OK.

Інші діагностичні команди (часто стають у нагоді)

# статус ноди (uptime, memory, queues, consumers)
docker compose exec rabbitmq rabbitmq-diagnostics status

# coverage check всіх listeners (порти 5672, 15672, тощо)
docker compose exec rabbitmq rabbitmq-diagnostics check_port_listener 5672

# health check (комплексний - listeners + ack-tracking + alarms)
docker compose exec rabbitmq rabbitmq-diagnostics -q check_running

# overview через CLI rabbitmqctl
docker compose exec rabbitmq rabbitmqctl list_queues
docker compose exec rabbitmq rabbitmqctl list_exchanges
docker compose exec rabbitmq rabbitmqctl list_bindings

RabbitMQ - мульти-протокольний broker. Кожен протокол слухає на окремому порту. У нашому sandbox експонуємо три з них.

docker compose exec rabbitmq rabbitmq-diagnostics listeners

URL і credentials

• URL: http://localhost:15672 (порт експоновано в compose.yaml).
• Дефолтний user: guest / guest з повним доступом до vhost /.

Обмеження guest user'а

З не-localhost log на broker'і покаже:

PLAIN login refused: user 'guest' can only connect via localhost

Чому це важливо. Якщо ви розгортаєте sandbox на віддаленому сервері і не можете залогінитись через UI - це не баг, це security default. Рішення:

1. Створити нового користувача з паролем (production-style):

docker compose exec rabbitmq rabbitmqctl add_user me <strong-password>
docker compose exec rabbitmq rabbitmqctl set_user_tags me administrator
docker compose exec rabbitmq rabbitmqctl set_permissions -p / me ".*" ".*" ".*"

2. Або (тільки для dev) - дозволити guest з remote: env-змінна RABBITMQ_LOOPBACK_USERS="" у compose.yaml. Не для production.

Цитати з офіційної сторінки docs/management. Згруповано за функцією.

Адміністрування

• "Declare, list and delete exchanges, queues, bindings, users, virtual hosts and user permissions"
• "Manage users", "Manage policies and runtime parameters"
• "Force close client connections, purge queues"

Моніторинг

• "Monitor queue length, message rates (globally and per queue, exchange or channel)"
• "Monitor node resource use: sockets and file descriptors, memory usage breakdown, available disk space"
• "View other users's connections and channels"

Дані

• "Export schema (vhosts, users, permissions, queues, exchanges, bindings, parameters, policies)" - експорт топології як JSON, можна імпортувати в інший broker.
• "Send and receive messages (useful in development environments and for troubleshooting)" - тестова публікація і Get-message прямо з UI.

Стандартні tabs у UI (станом на 4.x)

• Overview - графіки rates, totals, ноди кластера.
• Connections - відкриті TCP-з'єднання, peer host:port, idle/active, connection close.
• Channels - logical channels всередині кожного connection, prefetch, unacked.
• Exchanges - список exchanges, тип, durable/auto-delete; кнопка Add new exchange.
• Queues and Streams - queues, type (classic/quorum/stream), depth, consumers, rates.
• Admin - users, vhosts, policies, federation/shovel.

Мета: побачити primitives з 1.2 у живому UI і отримати feel за них. Без коду.

Крок 1: declare exchange

1. Exchanges tab → Add a new exchange.
2. Name: demo.events. Type: fanout. Durable: ✓. Auto delete: ✗.
3. Add exchange. Має з'явитись у списку.

Крок 2: declare queues

1. Queues tab → Add a new queue.
2. Type: Classic (для цього прикладу - але production-default уже Quorum; туди прийдемо в розділі 4).
3. Name: demo.email. Durable. Add queue.
4. Повторити для demo.analytics.

Крок 3: bind queues to exchange

1. Відкрити exchange demo.events → секція Bindings → Add binding from this exchange.
2. To queue: demo.email. Routing key порожній (fanout ігнорує). Bind.
3. Повторити для demo.analytics.

Крок 4: publish тестове повідомлення

1. На сторінці exchange demo.events → секція Publish message.
2. Routing key: будь-який (ігнорується для fanout). Payload: {"event": "user_signed_up"}. Publish.
3. UI покаже "Message published".

Крок 5: подивитися де воно

1. Queues tab → видно що demo.email і demo.analytics мають Ready: 1 (по копії в кожній - типовий fanout).
2. Відкрити queue → Get messages → Get Message(s). Побачити payload.

Закріплення 1.2.1a (properties) і 1.2.1b (types) на живому broker'і. Усі дії - через management UI на http://localhost:15672.

Крок 1: оголосити три queues з різними прапорами

На Queues and Streams tab → Add a new queue. Створи по черзі:

Важливо: для demo.transient (classic + non-durable + auto-delete без exclusive) можеш отримати помилку "queue declaration not allowed" у 4.3 - це той самий deprecated_features з 1.2.1a. Якщо так - просто пропусти або змінь на classic + durable.

Крок 2: подивись що видно у UI

Відкрий кожну queue по черзі. У секції Details:

• Type, Features (D = durable, AD = auto-delete, тощо).
• State (running / idle).
• Consumers - 0 на старті.
• Messages - 0.
• Arguments - те, що додав (x-max-length, x-queue-type).

Крок 3: тест auto-delete

1. Створи fanout-exchange demo.test і binding'и до всіх трьох queues.
2. Publish тестове повідомлення з UI у demo.test exchange. Усі три queues отримують копію (depth = 1).
3. Відкрий demo.transient → Get messages → Get Message(s) (натисни кілька разів).
4. Це не симулює "consumer subscribed/unsubscribed" - get-mode не лічиться. Auto-delete не спрацює.
5. Щоб реально побачити auto-delete - запусти PHP basic_consume на цю queue (з 2.1), потім Ctrl+C. Queue зникне з UI.

Крок 4: тест x-max-length + reject-publish

1. На сторінці demo.bounded exchange → Publish message. Опублікуй 6 повідомлень підряд.
2. Перші 5 - депт стає 5.
3. 6-те (без publisher confirms) - тихо дропнеться. З publisher confirms - повернеться basic.nack.
4. Перевір у demo.bounded → Messages = 5, не більше. Це і є back-pressure про який питав вище.

Cleanup

docker compose exec rabbitmq rabbitmqctl delete_queue demo.production
docker compose exec rabbitmq rabbitmqctl delete_queue demo.bounded
# demo.transient або зникне сама, або:
docker compose exec rabbitmq rabbitmqctl delete_queue demo.transient
docker compose exec rabbitmq rabbitmqctl delete_exchange demo.test

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare(
    'hello',                              // name
    false,                                // passive
    true,                                 // durable
    false,                                // exclusive
    false,                                // auto_delete
    false,                                // nowait
    new AMQPTable(['x-queue-type' => 'quorum'])  // arguments
);

$msg = new AMQPMessage('Hello World!');
$channel->basic_publish($msg, '', 'hello');   // exchange="", routing_key="hello"
echo " [x] Sent 'Hello World!'\n";

$channel->close();
$connection->close();

require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('hello', false, true, false, false, false,
    new AMQPTable(['x-queue-type' => 'quorum']));

echo " [*] Waiting for messages. To exit press CTRL+C\n";

$callback = function (AMQPMessage $msg) {
    echo ' [x] Received ', $msg->getBody(), "\n";
};

$channel->basic_consume(
    'hello',     // queue
    '',          // consumer_tag (auto)
    false,       // no_local
    true,        // no_ack ⚠ auto-ack (продакшн: false)
    false,       // exclusive
    false,       // nowait
    $callback    // callback
);

try {
    $channel->consume();
} catch (\Throwable $e) {
    echo $e->getMessage();
}

Кілька consumer'ів підписуються на ту саму queue → broker роздає повідомлення по черзі.

Code не потребує спецналаштувань - просто запусти кілька receive.php з тим самим queue.

Ack (acknowledgement, підтвердження) - сигнал від consumer'а до broker'а: "повідомлення оброблено успішно, можна стерти". Broker тримає повідомлення in-flight (на льоту) до отримання ack.

$callback = function (AMQPMessage $msg) {
    echo ' [x] Received ', $msg->getBody(), "\n";
    sleep(substr_count($msg->getBody(), '.')); // імітація роботи
    echo " [x] Done\n";
    $msg->ack();   // ← підтвердження після успішної обробки
};

$channel->basic_consume(
    'task_queue',
    '',
    false,
    false,    // no_ack=false → manual ack обов'язковий
    false,
    false,
    $callback
);

Що буває без ack

• Worker отримав m1, почав обробляти, впав → broker через connection-close re-queue m1 → інший worker отримає m1.
• Worker завершив, але не послав ack → broker думає що повідомлення in-flight, тримає його у пам'яті, не видаляє. Утечка memory.

Три способи відмови від обробки

Round-robin сліпий. Рішення: сказати broker'у "не давай worker'у наступного, поки попереднє не ack'ed".

$channel->basic_qos(
    null,    // prefetch_size (bytes; null = unlimited)
    1,       // prefetch_count = скільки in-flight на consumer'а
    false    // global (per-channel vs per-consumer)
);

Як це змінює поведінку

• prefetch_count=1 - найсуворіший: worker тримає максимум 1 unacked. Сповільнює, бо більше round-trip'ів broker↔worker.
• prefetch_count=10-50 - типове production-значення для CPU-bound worker'а: дозволяє pipeline без надлишкового memory.
• prefetch_count=1000+ - для дуже легких задач (millions/sec), щоб не упертись у round-trip latency.

Залежність від worker speed

Дві НЕЗАЛЕЖНІ опції, які мусять бути ОБИДВІ ввімкнені для гарантії "повідомлення переживе рестарт broker'а".

Durable queue

$channel->queue_declare(
    'task_queue',
    false,     // passive
    true,      // durable ← переживає рестарт broker'а
    false,     // exclusive
    false,     // auto_delete
    false,
    new AMQPTable(['x-queue-type' => 'quorum'])
);

Без durable=true queue видаляється при рестарті broker'а - разом з усіма повідомленнями.

Persistent message

$msg = new AMQPMessage($data, [
    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,  // = 2
]);
$channel->basic_publish($msg, '', 'task_queue');

Без delivery_mode=2 повідомлення сидить тільки у RAM. Broker впав → повідомлення зникло, навіть якщо queue durable.

Матриця гарантій

Сценарій: один producer кидає event - кожен consumer отримує копію (broadcast). Реалізація - fanout exchange + temporary exclusive queues.

Producer

$channel->exchange_declare(
    'logs',     // name
    'fanout',   // type
    false,      // passive
    false,      // durable
    false       // auto_delete
);

$msg = new AMQPMessage($data);
$channel->basic_publish($msg, 'logs');
// routing_key не вказаний → fanout все одно ігнорує

Consumer (логіка одноразової підписки)

$channel->exchange_declare('logs', 'fanout', false, false, false);

// queue з порожнім ім'ям → broker згенерує унікальне (amq.gen-XYZ)
// exclusive=true → видалиться при disconnect
list($queue_name, ,) = $channel->queue_declare(
    "",      // name (auto-generate)
    false,   // passive
    false,   // durable
    true,    // exclusive ← тільки для цього connection
    false    // auto_delete
);

$channel->queue_bind($queue_name, 'logs');

$channel->basic_consume($queue_name, '', false, true, false, false, $callback);

Чому exclusive queue, а не named

• Fresh empty: новий subscriber не повинен отримати повідомлення, які були до його старту.
• Auto-cleanup: consumer закрив connection → queue зникає, broker не зберігає мертві подвиги.

Класичний приклад tutorial - логи з рівнями info, warning, error. Consumer обирає, які рівні слухати.

Exchange + публікація

$channel->exchange_declare('direct_logs', 'direct', false, false, false);

$msg = new AMQPMessage($data);
$channel->basic_publish($msg, 'direct_logs', $severity);
// $severity ∈ {'info', 'warning', 'error'}

Consumer з кількома bindings

$channel->exchange_declare('direct_logs', 'direct', false, false, false);

list($queue_name, ,) = $channel->queue_declare("", false, false, true, false);

// Кілька bindings на ту саму queue з РІЗНИМИ routing keys
foreach ($severities as $severity) {
    $channel->queue_bind($queue_name, 'direct_logs', $severity);
}

$channel->basic_consume($queue_name, '', false, true, false, false, $callback);

Use case з консолі

# Логер що пише ВСЕ у файл
php receive_logs_direct.php info warning error  > logs/all.log

# Окрема панель для критичних
php receive_logs_direct.php error  | mail -s "ERROR" oncall@example.com

Direct - точна рівність. Topic - patterns з wildcards. Routing key стає ієрархічним: <facility>.<severity> або <entity>.<action>.<region>.

$channel->exchange_declare('topic_logs', 'topic', false, false, false);

// Producer публікує з ієрархічним routing key
$channel->basic_publish($msg, 'topic_logs', 'kern.critical');
$channel->basic_publish($msg, 'topic_logs', 'auth.warning');
$channel->basic_publish($msg, 'topic_logs', 'kern.info');

// Consumer 1: всі kern-events
$channel->queue_bind($queueA, 'topic_logs', 'kern.*');

// Consumer 2: всі critical, незалежно від facility
$channel->queue_bind($queueB, 'topic_logs', '*.critical');

// Consumer 3: ВСЕ
$channel->queue_bind($queueC, 'topic_logs', '#');

Wildcards (нагадування з 1.2.5)

• * - рівно одне слово.
• # - нуль або більше слів.
• Слова відокремлюються крапкою. Routing key до 255 байт.

Сценарій з курсу: e-commerce events

// Producer - кожна доменна подія публікується з ієрархічним key
publish($msg, 'events', 'order.created.eu');
publish($msg, 'events', 'order.paid.us');
publish($msg, 'events', 'user.signed_up.eu');

// Consumer'и за зоною відповідальності:
queue_bind($euAnalyticsQ, 'events', '*.*.eu');         // тільки EU events
queue_bind($paymentsQ,    'events', 'order.paid.*');   // тільки payments
queue_bind($auditQ,       'events', '#');              // ВСЕ для аудиту

Питання, що вирішується: як producer'у дізнатись, що broker реально прийняв повідомлення? Без додаткових механізмів basic_publish - "fire-and-forget": producer запхав у TCP-сокет, broker міг впасти між publish і flush на диск.

Два механізми

Як вмикати в php-amqplib

$channel->confirm_select();   // confirm.select - переводить channel у confirm mode

// Реєструємо handlers до ack/nack від broker'а
$channel->set_ack_handler(function (AMQPMessage $msg) {
    // повідомлення з sequence number $msg->getDeliveryTag() прийнято
});
$channel->set_nack_handler(function (AMQPMessage $msg) {
    // broker відмовився - log, retry, alert
});

for ($i = 0; $i < 1000; $i++) {
    $channel->basic_publish(new AMQPMessage("msg $i"), '', 'tasks');
}

// Чекати, поки прийдуть всі ack'и (або timeout 5 сек)
$channel->wait_for_pending_acks_returns(5);

Що буде, якщо publish у exchange без bindings (або routing key не матчить жодну binding)? За замовчуванням - тиха втрата ("zero or more queues" з 1.2.1).

Прапор mandatory

$msg = new AMQPMessage('payload');

$channel->basic_publish(
    $msg,
    'orders',
    'unmatched.key',
    true   // mandatory = true ← вимагати, щоб повідомлення потрапило хоча б в одну queue
);

$channel->set_return_listener(function ($replyCode, $replyText, $exchange, $routingKey, $msg) {
    // повідомлення повернуто broker'ом, бо нікуди не маршрутизувалось
    error_log("Returned: $replyText for $exchange/$routingKey");
});

Чому це важливо

• У production легко "забути" створити binding після додавання нового consumer'а - повідомлення тихо губляться.
• Mandatory + return listener перетворюють "тиху втрату" на гучну (log, alert).
• Альтернатива: alternate exchange (x-alternate-exchange у exchange-args) - перенаправляє unrouted повідомлення в окрему "catch-all" чергу. Менший overhead, ніж listener у кожному producer'і.

Базовий manual ack-flow вже знайомий з 2.2.2. У RabbitMQ 4.3 - значуща зміна щодо acknowledgement timeout.

Що таке consumer ack timeout

• Default value: 30 minutes ("The default timeout value for RabbitMQ is 30 minutes").
• Якщо consumer мовчить (не ack/nack) довше за timeout - broker закриває channel помилкою:

Consumer 'consumer-tag-998754663370' on channel 1 and queue 'qq.1' in vhost '/'
has timed out waiting for a consumer acknowledgement of a delivery with delivery tag = 10.
Timeout used: 1800000 ms.

Як налаштувати

# /etc/rabbitmq/rabbitmq.conf (мілісекунди)
consumer_timeout = 1800000   # 30 хв (default)
# або 1 година для довгих ETL-задач:
consumer_timeout = 3600000

Practical impact для backend-команди

• Worker, що робить тривалу задачу (відеоконвертація, великий ETL) має або: (а) ack одразу і записати state у DB як "in progress" (idempotent recovery), або (б) збільшити consumer_timeout, або (в) розбити задачу на чанки з ack після кожного.
• Без ack у відведений час - broker re-queue'ить повідомлення, інший worker його обробить → можливі дублікати (тема 3.4).

Чотири причини dead-letter

Що таке "стало мертвим"

Broker re-publish'є його у Dead Letter Exchange (DLX) з:

• оригінальним body;
• headers x-death - історія, чому повідомлення стало dead (queue, reason, count, time);
• routing key = original або x-dead-letter-routing-key, якщо налаштовано.

Базове налаштування queue

$channel->exchange_declare('dlx', 'direct', false, true, false);
$channel->queue_declare('dead_letters', false, true, false, false);
$channel->queue_bind('dead_letters', 'dlx', 'failed');

$channel->queue_declare('tasks', false, true, false, false, false, new AMQPTable([
    'x-queue-type' => 'quorum',
    'x-dead-letter-exchange'    => 'dlx',
    'x-dead-letter-routing-key' => 'failed',
]));

Retry з exponential backoff (через TTL + DLX)

Класичний патерн: повідомлення впало → іде в "park"-queue з TTL → expire → DLX повертає назад у головну queue. Між спробами - exponential delay.

// Дві queues:
// 1) main "tasks" - звичайна
// 2) "tasks.retry.5s" з x-message-ttl=5000 і DLX = main exchange

$channel->queue_declare('tasks.retry.5s', false, true, false, false, false, new AMQPTable([
    'x-message-ttl'             => 5000,         // 5 секунд
    'x-dead-letter-exchange'    => '',           // default exchange
    'x-dead-letter-routing-key' => 'tasks',      // повертає в main queue
]));

// Consumer при тимчасовій помилці:
function onTransientError(AMQPMessage $msg, $channel) {
    // Re-publish у retry queue замість nack
    $channel->basic_publish($msg, '', 'tasks.retry.5s');
    $msg->ack();   // ack оригіналу
}

Patterns на DLX

• Parking lot: один dead_letters queue для аудиту. Inспектор переглядає, можливо ручне re-publish.
• Exponential backoff: кілька retry queues з різним TTL: retry.1s, retry.5s, retry.30s, retry.5m. Лічильник - у header'і повідомлення.
• Native delayed retry (4.3): quorum queues підтримують x-delivery-limit + x-dead-letter-strategy: at-least-once. Простіше, ніж параметр.

Idempotent (ідемпотентна) операція

Операція, повторне виконання якої не змінює результат після першого. Формально: f(f(x)) == f(x).

Дві стратегії на consumer-side

1. Idempotent by design

Спроектувати операцію так, щоб повтор був безпечний. Useful patterns:

• UPSERT (PostgreSQL ON CONFLICT DO NOTHING/UPDATE) замість INSERT.
• SET state = 'paid' замість UPDATE state = state | PAID_FLAG.
• Перевірка стану перед дією: "чи це вже виконано?".

2. Deduplication через зовнішній storage

$messageId = $msg->get('message_id');  // або власний idempotency key у payload

// Redis SETNX (set if not exists) з TTL
$inserted = $redis->set("processed:$messageId", "1", ['EX' => 86400, 'NX' => true]);

if ($inserted) {
    // Перша спроба - обробити
    handleMessage($msg);
    $msg->ack();
} else {
    // Дублікат - просто ack без обробки
    $msg->ack();
}

Як працює реплікація

"All operations (state changes) on a quorum queue are sent to a primary member called a leader which in turn replicates the operations to the remaining members, called followers."

• Raft - алгоритм консенсусу (як Paxos, але простіший в реалізації). Гарантує: якщо більшість (quorum) нод погодилась записати, дані не загубляться.
• Leader / Follower - на кожну queue є один leader, кілька followers. Producer пише leader, leader реплікує followers, дочекавшись ACK від quorum (більшість) - відповідає producer'у.
• Auto-failover: якщо leader ноду губиться, followers вибирають нового leader'а через Raft-голосування.

Декларування

$channel->queue_declare('orders', false, true, false, false, false, new AMQPTable([
    'x-queue-type' => 'quorum',
    // 'x-quorum-initial-group-size' => 3, // default 3
    // 'x-delivery-limit' => 10,           // після 10 redelivers → DLX
]));

"To declare a quorum queue set the x-queue-type queue argument to quorum (the default is classic)."

Розмір кластера

• 3 ноди - офіц. рекомендований default. Витримує падіння 1 ноди.
• 5 нод - для критичних navrh. Витримує падіння 2 нод.
• "performance tails off quite a bit for quorum queue node sizes larger than 5. We do not recommend running quorum queues on more than 7 RabbitMQ nodes".

З документації - чотири категорії, де quorum queues неефективні або непридатні:

Як вирішити: classic, quorum чи stream

Версія 4.3.0 принесла кілька значущих покращень саме для quorum queues. Цитати з release notes.

1. Strict priority queues

• Раніше priority queues були тільки classic. У 4.3 - quorum.
• x-max-priority: N при declare. Producer вказує priority: K у message properties.
• Strict = вищий priority обробляється спочатку. Не "вища ймовірність", саме black-and-white.

2. Delayed retry з configurable backoff

• Раніше для retry-with-backoff потрібно було будувати ланцюг TTL+DLX (як у 3.3.2). У 4.3 quorum queue робить це нативно.
• Налаштовується через x-overflow: reject-publish + x-delivery-limit + новий x-dead-letter-strategy.

3. Configurable consumer timeout per queue

• Раніше - тільки global config. У 4.3 - x-consumer-timeout: 1800000 per-queue.
• Корисно: довгі ETL queues - 1 година, швидкі notifications - 10 секунд.

4. Memory optimization

Для systems з мільйонами queued messages - до 50% memory зменшення на broker'і.

Ключова відмінність від queues

Декларування

$channel->queue_declare('events', false, true, false, false, false, new AMQPTable([
    'x-queue-type' => 'stream',
    'x-max-age'    => '7D',         // зберігати 7 днів
    'x-stream-max-segment-size-bytes' => 100_000_000, // 100MB segments
]));

"set the x-queue-type queue argument to stream ... must be provided by a client at declaration time; it cannot be set or changed using a policy."

Use cases

• Event sourcing: кожен domain event у stream, materialized views читають з offset 0 при старті.
• Audit log: compliance вимагає replay - stream не дає видалити випадково.
• Fan-out з replay: новий consumer (наприклад analytics-сервіс) робить read-from-beginning без впливу на існуючих.
• Високий throughput: sequential append на диск - набагато швидший за random write.

Stream доступний по двох протоколах одночасно:

PHP сценарій (через AMQP 0-9-1)

// Декларуємо stream
$channel->queue_declare('events', false, true, false, false, false, new AMQPTable([
    'x-queue-type' => 'stream',
]));

// Consumer з offset = "first" (з самого початку)
$channel->basic_qos(null, 100, false);  // prefetch обов'язковий для streams
$channel->basic_consume(
    'events',
    '',
    false,
    false,   // manual ack
    false,
    false,
    $callback,
    null,
    new AMQPTable([
        'x-stream-offset' => 'first',  // або: 'last', 'next', timestamp, або int offset
    ])
);

Offset стратегії

• 'first' - з самого початку (replay all).
• 'last' - з останнього chunk'а (catch-up).
• 'next' - тільки нові повідомлення.
• Timestamp - з конкретного моменту.
• Numeric offset - з конкретного зміщення.

Чому "pure PHP"

• Не вимагає PECL-extension. Інша бібліотека (php-amqp) вимагає ext-amqp з C-кодом - складно ставити у Docker, конфлікти з PHP-версіями.
• php-amqplib працює на будь-якому PHP-хості.
• Trade-off: повільніше за C-extension у high-throughput сценаріях (~10-30%).

Інсталяція

composer require php-amqplib/php-amqplib

Базовий flow

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

// 1. Connection (TCP) - один раз на процес
$connection = new AMQPStreamConnection(
    'rabbitmq-host', 5672, 'user', 'password',
    '/',                  // vhost
    false,                // insist
    'AMQPLAIN',           // login_method
    null,                 // login_response
    'en_US',              // locale
    3.0,                  // connection_timeout
    3.0,                  // read_write_timeout
    null,                 // context (TLS)
    false,                // keepalive
    60,                   // heartbeat (секунди!)
    0.0,                  // channel_rpc_timeout
);

// 2. Channel - дешевий, можна багато на 1 connection
$channel = $connection->channel();

// 3. Декларуй topology idempotently
$channel->exchange_declare('orders', 'topic', false, true, false);
$channel->queue_declare('order_processing', false, true, false, false, false,
    new AMQPTable(['x-queue-type' => 'quorum']));
$channel->queue_bind('order_processing', 'orders', 'order.*.processing');

// 4. Publish (з headers і properties)
$msg = new AMQPMessage(json_encode($payload), [
    'content_type'   => 'application/json',
    'delivery_mode'  => AMQPMessage::DELIVERY_MODE_PERSISTENT,
    'message_id'     => Uuid::v4()->toRfc4122(),
    'timestamp'      => time(),
    'application_headers' => new AMQPTable(['source' => 'orders-service']),
]);
$channel->basic_publish($msg, 'orders', 'order.created.eu');

// 5. Завжди закривати в кінці процесу
$channel->close();
$connection->close();

Дві варіанти Connection класу

• AMQPStreamConnection - PHP streams API. Default. Підтримує TLS через context.
• AMQPSocketConnection - low-level socket_* функції. Швидше, але потребує ext-sockets. Без TLS support.
• AMQPSSLConnection - спеціалізований для TLS поверх stream connection.

Інсталяція (зверни увагу: PECL-extension!)

composer require symfony/amqp-messenger
# І в Docker - PECL extension:
# docker-php-ext-install ... && pecl install amqp && docker-php-ext-enable amqp

DSN format

# .env
MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
#                        scheme   user   pass   host  port vhost  exchange
#                        amqps:// для TLS

%2f - це URL-encoded / (default vhost).

Конфігурація з retry strategy

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    auto_setup: true   # автоматично створює exchange/queue/bindings
                    exchange:
                        name: 'orders'
                        type: 'topic'
                    queues:
                        order_processing:
                            arguments:
                                x-queue-type: 'quorum'
                            binding_keys: ['order.*.processing']
                retry_strategy:
                    max_retries: 3
                    delay: 1000        # ms перед першим retry
                    multiplier: 2      # exponential: 1s, 2s, 4s
                    max_delay: 10000
                    jitter: 0.1        # ±10% randomness
        routing:
            App\Message\OrderCreatedEvent: async

auto_setup: "By default, the transport will automatically create any exchanges, queues and binding keys that are needed. That can be disabled." У production часто auto_setup: false + topology через migration.

Worker

# Запуск consumer'а (виконується infinitely до Ctrl+C / SIGTERM)
php bin/console messenger:consume async

# З verbose logging
php bin/console messenger:consume async -vv

# З time limit (для systemd / supervisord auto-restart)
php bin/console messenger:consume async --time-limit=3600

# Зупинити gracefully
php bin/console messenger:stop-workers

Community-стандарт: vladimir-yuldashev/laravel-queue-rabbitmq. Drop-in заміна Redis/SQS - вся Queue-фасадна логіка Laravel працює без змін.

Інсталяція

composer require vladimir-yuldashev/laravel-queue-rabbitmq
# Service provider реєструється автоматично через Laravel Package Discovery

Конфігурація

// config/queue.php
'default' => env('QUEUE_CONNECTION', 'rabbitmq'),

'connections' => [
    'rabbitmq' => [
        'driver' => 'rabbitmq',
        'queue'  => env('RABBITMQ_QUEUE', 'default'),
        'hosts' => [
            [
                'host'     => env('RABBITMQ_HOST', '127.0.0.1'),
                'port'     => env('RABBITMQ_PORT', 5672),
                'user'     => env('RABBITMQ_USER', 'guest'),
                'password' => env('RABBITMQ_PASSWORD', 'guest'),
                'vhost'    => env('RABBITMQ_VHOST', '/'),
            ],
        ],
        'options' => [
            'queue' => [
                'job'         => \VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob::class,
                'prefetch'    => 10,
                'declare'     => true,    // auto-declare queue
                'arguments'   => [
                    'x-queue-type' => 'quorum',
                ],
            ],
        ],
    ],
],

Використання (стандартне Laravel API)

// Job клас не змінюється - той самий що для Redis/SQS
class ProcessOrder implements ShouldQueue {
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public function __construct(public Order $order) {}

    public function handle(): void {
        // ... обробка
    }
}

// Dispatch як зазвичай
ProcessOrder::dispatch($order)->onQueue('orders.processing');

Дві команди для consumer'а

Гібрид у production

У реальних проєктах часто комбінують:

• Symfony Messenger для domain events (decoupling сервісів) - UserSignedUp, OrderPaid.
• php-amqplib у окремому worker'і для streams з replay (analytics, audit).
• Laravel Queue якщо легасі частина в Laravel - щоб не писати їх своїм Messenger.

П'ять discovery механізмів

Чому 3 ноди (а не 2 чи 1)

Базується на правилі quorum = majority для Raft (Quorum Queues) і Khepri (metadata store).

Створи окремий compose.cluster.yaml у sandbox/ для практики. Базовий каркас:

x-rabbit-base: &rabbit-base
  image: rabbitmq:4.3-management
  environment: &rabbit-env
    RABBITMQ_ERLANG_COOKIE: 'study-cluster-cookie'
    RABBITMQ_DEFAULT_USER: guest
    RABBITMQ_DEFAULT_PASS: guest
  healthcheck:
    test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
    interval: 15s
    timeout: 10s
    retries: 5

services:
  rabbit1:
    <<: *rabbit-base
    hostname: rabbit1
    container_name: rabbit1
    ports:
      - "5672:5672"
      - "15672:15672"
    volumes:
      - rabbit1-data:/var/lib/rabbitmq
      - ./cluster-config/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro

  rabbit2:
    <<: *rabbit-base
    hostname: rabbit2
    container_name: rabbit2
    volumes:
      - rabbit2-data:/var/lib/rabbitmq
      - ./cluster-config/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro

  rabbit3:
    <<: *rabbit-base
    hostname: rabbit3
    container_name: rabbit3
    volumes:
      - rabbit3-data:/var/lib/rabbitmq
      - ./cluster-config/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro

volumes:
  rabbit1-data:
  rabbit2-data:
  rabbit3-data:

cluster-config/rabbitmq.conf

cluster_formation.peer_discovery_backend = classic_config
cluster_formation.classic_config.nodes.1 = rabbit@rabbit1
cluster_formation.classic_config.nodes.2 = rabbit@rabbit2
cluster_formation.classic_config.nodes.3 = rabbit@rabbit3

# Khepri за замовчуванням у 4.3
# log і management plugin вже включені у -management образі

# Disk alarm
disk_free_limit.absolute = 2GB

# Memory alarm
vm_memory_high_watermark.relative = 0.6

Перевірка кластера

docker compose -f compose.cluster.yaml up -d

# Дочекатися healthy всіх 3 нод (60-90 с)
docker compose -f compose.cluster.yaml ps

# Перевірити cluster status з будь-якої ноди
docker exec rabbit1 rabbitmqctl cluster_status
# Має показати 3 nodes: [rabbit@rabbit1, rabbit@rabbit2, rabbit@rabbit3]

# Створити quorum queue з 3-way replication
docker exec rabbit1 rabbitmqadmin declare queue name=test \
    arguments='{"x-queue-type":"quorum"}'

Окрім UI, management plugin експонує REST API на тому ж порту 15672. Все що видно в UI, можна скачати JSON'ом.

Ключові endpoint'и

Приклад: швидкий health-check скрипт

#!/bin/bash
# health.sh
HOST="http://guest:guest@localhost:15672"

# Перевіряємо: всі ноди running?
unhealthy=$(curl -s "$HOST/api/nodes" | jq '[.[] | select(.running == false)] | length')
[[ "$unhealthy" -gt 0 ]] && echo "FAIL: $unhealthy node(s) not running" && exit 1

# Перевіряємо: queues з depth > 10000?
backed_up=$(curl -s "$HOST/api/queues" \
    | jq '[.[] | select(.messages > 10000)] | length')
[[ "$backed_up" -gt 0 ]] && echo "WARN: $backed_up queue(s) backed up"

# Memory alarm?
mem_alarm=$(curl -s "$HOST/api/overview" | jq '.message_stats // {}')
echo "OK"

Ключові метрики для backend-команди

• Queue depth (messages + messages_unacknowledged) - чи consumers встигають.
• Publish/deliver rate - throughput; різке падіння → щось зламалось.
• Connection churn (відкриття/закриття на хвилину) - якщо worker reconnect'ться - неправильний lifecycle.
• Memory used vs limit - якщо >80% від vm_memory_high_watermark, broker почне throttle producers.
• FD usage vs limit - кожне з'єднання + queue + open file - 1 FD.

Включений за замовчуванням у -management tag

На наших sandbox-нодах він уже працює - це той самий порт 15692, який ми експонуємо у compose.yaml.

# Перевірити
curl -s localhost:15692/metrics | head -n 20

# rabbitmq_build_info{erlang_version="26.2.5",rabbitmq_version="4.3.0"} 1
# rabbitmq_identity_info{...} 1
# rabbitmq_alarms_memory_used_watermark 0
# rabbitmq_alarms_disk_limit_exceeded 0
# rabbitmq_queue_messages 0
# ...

Якщо вимкнено - як включити

rabbitmq-plugins enable rabbitmq_prometheus

Два режими metrics

• Aggregated (default): "metrics are aggregated by name. This mode has lower performance overhead with the output size constant, even as the number of objects grows." Один rabbitmq_queue_messages показує total по всіх queues.
• Per-object: "individual metric for each object-metric pair. With a large number of stats-emitting entities, this can result in very large payloads." Окремий rabbitmq_queue_messages{queue="orders"} на кожну queue. Корисно для drill-down, але дорого при тисячах queues.

# Per-object endpoint
curl -s localhost:15692/metrics/per-object

Топ-7 алертів (PromQL)

# 1. Нода не відповідає 2 хв
up{job="rabbitmq"} == 0  for 2m

# 2. Memory alarm активний
rabbitmq_alarms_memory_used_watermark == 1

# 3. Disk alarm активний
rabbitmq_alarms_disk_limit_exceeded == 1

# 4. Queue depth перевищив поріг
rabbitmq_queue_messages > 10000

# 5. Queue без consumers
rabbitmq_queue_consumers == 0  and  rabbitmq_queue_messages > 100

# 6. Connection churn (rate of new connections / sec)
rate(rabbitmq_connections_opened_total[5m]) > 10

# 7. FD usage > 80%
rabbitmq_process_open_fds / rabbitmq_process_max_fds > 0.8

Скорочена вибірка з офіційного production-checklist, з акцентом на пункти, які напряму зачіпають backend-команду.

Operating system

• File descriptors: "at least 50K open file descriptors for the effective RabbitMQ user". Кожен connection + queue + open file = FD.
• Перевірка: cat /proc/$(pidof beam.smp)/limits | grep "open files".
• Storage: "Modern RabbitMQ features, such as Khepri, quorum queues and streams, are designed for durable storage only." NFS / Ceph / GlusterFS - заборонені. Local SSD / EBS gp3.

Resource alarms

• Memory: default vm_memory_high_watermark.relative = 0.6 (60%). При перевищенні broker блокує publishers (back-pressure). "keep the relative watermark between 0.4-0.7".
• Disk: disk_free_limit.absolute = 2GB (мінімум). "all disk writes will fail" при перевищенні - дані в небезпеці.

Security

• Видалити default user: "For production environments, delete the default user (guest)". Створити іменованих з мінімальними permissions.
• TLS: "We recommend using TLS connections when possible, at least to encrypt traffic". Сертифікати на 5671 (AMQPS), 15671 (Management HTTPS), 15691 (Prometheus HTTPS).
• Vhosts: один vhost на кожен tenant / середовище. Permissions per-vhost.
• Inter-node ports: 4369 (epmd), 25672 (Erlang dist) - закрити від external traffic, дозволити тільки між нодами.

Erlang OTP

• Вимога RabbitMQ 4.3: Erlang/OTP 26.2+, рекомендовано 27.x.
• Однакова версія на всіх нодах кластера. Mismatched OTP - ризик невидимих compatibility-проблем.

Cluster topology

• "an odd number of nodes (3, 5, 7, etc)".
• Partition handling: cluster_partition_handling = pause_minority - меншина при split-brain паузить себе, чекає на возз'єднання.
• Backups: топологію rabbitmqctl export_definitions /tmp/defs.json - users, vhosts, exchanges, bindings, policies. Зберігати у git або S3.