Урок 04

RabbitMQ

Что такое RabbitMQ

RabbitMQ — брокер сообщений с расширенной маршрутизацией.

Отличия от Redis:

Надёжность — сообщения сохраняются на диск.
Гарантии — подтверждения доставки.
Маршрутизация — exchanges, routing keys.
Dead letter — встроенная очередь для failed.
Приоритеты — на уровне очереди.

Когда Redis лучше:

Простые очереди.
Максимальная скорость.
Не нужна сложная маршрутизация.

Когда RabbitMQ лучше:

Критичные сообщения.
Сложная маршрутизация.
RPC между сервисами.
Микросервисы.

Установка

Docker:

docker run -d --name rabbitmq \
    -p 5672:5672 -p 15672:15672 \
    rabbitmq:3-management

URL UI: http://localhost:15672 (guest/guest).

Основные концепции

Producer — отправляет сообщение в exchange.

Exchange — маршрутизирует в очереди.

Queue — хранит сообщения.

Consumer — читает из очереди.

Binding — связь exchange → queue.

Producer → Exchange → [Binding] → Queue → Consumer

Типы Exchange

Direct — точное совпадение routing key:

exchange: orders
exchange orders → routing key 'order.created' → queue order-emails

Fanout — всем очередям:

exchange logs → queue analytics + queue audit + queue alerts

Topic — по шаблону:

exchange events → 'user.*' → queue user-events
exchange events → '*.created' → queue created-events

Headers — по заголовкам (редко).

Установка клиента

Node.js:

npm install amqplib

PHP:

composer require php-amqplib/php-amqplib

Producer (Node.js)

import amqp from 'amqplib';

const conn = await amqp.connect('amqp://guest:guest@localhost:5672');
const channel = await conn.createChannel();

const exchange = 'orders';
await channel.assertExchange(exchange, 'topic', { durable: true });

channel.publish(
    exchange,
    'order.created',
    Buffer.from(JSON.stringify({ orderId: 123, total: 500 })),
    { persistent: true },
);

console.log('Sent');

Что здесь:

assertExchange — создать если нет.
durable: true — переживёт рестарт.
persistent: true — сохранится на диск.

Consumer (Node.js)

const conn = await amqp.connect('amqp://guest:guest@localhost:5672');
const channel = await conn.createChannel();

const exchange = 'orders';
const queue = 'order-emails';

await channel.assertExchange(exchange, 'topic', { durable: true });
await channel.assertQueue(queue, { durable: true });
await channel.bindQueue(queue, exchange, 'order.created');

channel.prefetch(10);   // по 10 сообщений за раз

channel.consume(queue, async (msg) => {
    if (!msg) return;

    try {
        const data = JSON.parse(msg.content.toString());
        await sendOrderEmail(data);

        channel.ack(msg);   // подтверждение
    } catch (err) {
        console.error(err);
        channel.nack(msg, false, true);   // вернуть в очередь
    }
}, { noAck: false });

Что здесь:

prefetch(10) — не более 10 неподтверждённых.
ack — успех.
nack — вернуть или отбросить.

Fanout — всем

await channel.assertExchange('logs', 'fanout', { durable: true });

// Три очереди
await channel.assertQueue('logs:analytics', { durable: true });
await channel.assertQueue('logs:alerts', { durable: true });

// Без routing key
await channel.bindQueue('logs:analytics', 'logs', '');
await channel.bindQueue('logs:alerts', 'logs', '');

// Одно сообщение — всем
channel.publish('logs', '', Buffer.from('user registered'));

Что даёт: каждое сообщение — всем подписчикам.

Topic — по шаблону

await channel.assertExchange('events', 'topic', { durable: true });

await channel.assertQueue('user-events');
await channel.bindQueue('user-events', 'events', 'user.*');

await channel.assertQueue('all-created');
await channel.bindQueue('all-created', 'events', '*.created');

// Матчится user.*
channel.publish('events', 'user.created', ...);

// Матчится обоими
channel.publish('events', 'user.created', ...);

Шаблоны:

* — одно слово.
# — ноль или больше слов.
user.* — user.created, user.deleted.
order.# — order, order.created, order.created.paid.

RPC — запрос-ответ

Клиент:

const correlationId = crypto.randomUUID();
const { queue } = await channel.assertQueue('', { exclusive: true });

channel.sendToQueue(
    'rpc_queue',
    Buffer.from(JSON.stringify({ userId: 1 })),
    { correlationId, replyTo: queue },
);

channel.consume(queue, (msg) => {
    if (msg?.properties.correlationId === correlationId) {
        const result = JSON.parse(msg.content.toString());
        channel.ack(msg);
        console.log('Result:', result);
    }
}, { noAck: false });

Сервер:

channel.consume('rpc_queue', async (msg) => {
    if (!msg) return;

    const { userId } = JSON.parse(msg.content.toString());
    const result = await getUserTariff(userId);

    channel.sendToQueue(
        msg.properties.replyTo,
        Buffer.from(JSON.stringify(result)),
        { correlationId: msg.properties.correlationId },
    );

    channel.ack(msg);
});

Что даёт: синхронный вызов между сервисами через очередь.

Dead Letter Queue

Очередь с DLX:

await channel.assertQueue('orders', {
    durable: true,
    deadLetterExchange: 'orders.dlx',
    deadLetterRoutingKey: 'failed',
});

await channel.assertExchange('orders.dlx', 'direct', { durable: true });
await channel.assertQueue('orders.failed', { durable: true });
await channel.bindQueue('orders.failed', 'orders.dlx', 'failed');

Что даёт: все nack/reject идут в orders.failed.

TTL — время жизни

На сообщение:

channel.publish(exchange, routingKey, content, {
    expiration: '60000',   // 60 сек
});

На очередь:

await channel.assertQueue('temp', {
    durable: false,
    arguments: { 'x-message-ttl': 60000 },
});

Правило: expired сообщения → DLQ.

Приоритеты

await channel.assertQueue('tasks', {
    durable: true,
    maxPriority: 10,
});

channel.publish(exchange, key, content, { priority: 9 });   // важнее
channel.publish(exchange, key, content, { priority: 1 });   // обычное

Что даёт: приоритетные вперёд.

Lazy queues

Проблема: миллион сообщений жрёт RAM.

Решение: lazy — хранятся на диск.

await channel.assertQueue('bulk', {
    durable: true,
    arguments: { 'x-queue-mode': 'lazy' },
});

Что даёт: RAM почти не тратится.

Отличие от Laravel Queue

Laravel по умолчанию: Redis / database.

Laravel + RabbitMQ:

composer require vyuldashev/laravel-queue-rabbitmq

QUEUE_CONNECTION=rabbitmq
RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_QUEUE=default

// config/queue.php
'rabbitmq' => [
    'driver' => 'rabbitmq',
    'host' => env('RABBITMQ_HOST'),
    'port' => env('RABBITMQ_PORT'),
    'vhost' => env('RABBITMQ_VHOST', '/'),
    'login' => env('RABBITMQ_LOGIN'),
    'password' => env('RABBITMQ_PASSWORD'),
    'queue' => env('RABBITMQ_QUEUE', 'default'),
    'options' => [
        'queue' => [
            'exchange' => 'laravel',
            'exchange_type' => 'topic',
            'queue_max_priority' => 10,
        ],
    ],
],

Dispatch как обычно:

SendEmailJob::dispatch($user)->onQueue('emails');

Межъязыковой обмен

PHP → Node.js:

// Laravel публикует
$this->amqp->publish(
    'events',
    'user.created',
    json_encode(['userId' => $user->id]),
);

// Node.js читает
channel.consume('user-events', (msg) => {
    const { userId } = JSON.parse(msg.content.toString());
    // ...
});

Правило: JSON — универсальный формат.

Мониторинг

Management UI:

Queues — размер, consumers.
Messages — publish / deliver / ack.
Connections — кто подключён.
Channels — открытые каналы.

Метрики:

Messages ready — сколько ждёт.
Messages unacked — сколько в работе.
Message rate — сколько в секунду.
Consumer count — живые воркеры.

Практика: система событий

1. Exchange:

await channel.assertExchange('app-events', 'topic', { durable: true });

2. Producer (Laravel):

public function publish(string $event, array $data): void
{
    $this->amqp->publish('app-events', $event, json_encode($data));
}

// Использование
$this->publish('user.registered', ['userId' => $user->id]);
$this->publish('order.created', ['orderId' => $order->id]);

3. Consumers (Node.js):

// Аналитика
await channel.assertQueue('analytics');
await channel.bindQueue('analytics', 'app-events', '#');   // всё

// Email
await channel.assertQueue('email-service');
await channel.bindQueue('email-service', 'app-events', 'user.*');
await channel.bindQueue('email-service', 'app-events', 'order.created');

4. Обработка:

channel.consume('email-service', async (msg) => {
    const event = msg.fields.routingKey;
    const data = JSON.parse(msg.content.toString());

    switch (event) {
        case 'user.registered':
            await sendWelcome(data.userId);
            break;
        case 'order.created':
            await sendOrderConfirmation(data.orderId);
            break;
    }

    channel.ack(msg);
});

Частые ошибки

1. Забыли durable: true:

После рестарта очереди исчезли

2. Сообщения не persistent:

После рестарта задачи потерялись

Решение: { persistent: true } при publish.

3. Забыли ack:

Unacked растёт, память кончается

Решение: channel.ack(msg) всегда.

4. Нет prefetch:

Один воркер забирает все сообщения

Решение: channel.prefetch(10).

5. Consumer падает без nack:

Сообщение навсегда unacked

Решение: обернуть в try/catch + nack.

6. Один exchange на всё:

Маршрутизация превращается в кашу

Решение: по одному exchange на домен (orders, users, logs).

Итог

RabbitMQ — надёжный брокер с маршрутизацией.
Exchange типы: direct, fanout, topic, headers.
Producer → exchange → binding → queue → Consumer.
ack / nack — подтверждения.
prefetch — контроль нагрузки.
DLX — Dead Letter.
RPC — запрос-ответ через очередь.
Laravel — пакет laravel-queue-rabbitmq.
Management UI — :15672.

Дальше — real-time уведомления.