Подключение RabbitMQ

Внимание

Данная статья документации написана нейросетью. Возможны некоторые расхождения и неточность информации!

Руководство по подключению к RabbitMQ в IO7 фреймворке.

Содержание

Требования

Для работы с RabbitMQ необходим пакет php-amqplib/php-amqplib:

composer require php-amqplib/php-amqplib

RabbitMQManager

Пространство имён: IO\Queue

Менеджер для работы с RabbitMQ. Реализован как синглтон.

Инициализация

init()

public static function init(array $config): self

Инициализировать или получить существующий экземпляр.

ПараметрТипЗначение по умолчаниюОбязателенОписание
$configarray<string, mixed>—ДаКонфигурация подключения

Параметры конфигурации:

КлючЗначение по умолчаниюОбязателенОписание
host—ДаАдрес сервера
user—ДаПользователь
password—ДаПароль
port5672НетПорт сервера
vhost'/'НетВиртуальный хост
use IO\Queue\RabbitMQManager;

RabbitMQManager::init([
    'host'     => 'localhost',
    'port'     => 5672,
    'user'     => 'guest',
    'password' => 'guest',
    'vhost'    => '/',
]);

getInstance()

public static function getInstance(): self

Получить синглтон-экземпляр. Выбрасывает \Exception если не инициализирован.

$mq = RabbitMQManager::getInstance();

Публикация сообщений

publish()

public function publish(
    string $queue,
    $message,
    string $exchange = '',
    int $deliveryMode = 2
): bool

Отправить сообщение в очередь.

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаИмя очереди
$messagestring|array—ДаСообщение (массив сериализуется в JSON)
$exchangestring''НетИмя exchange (default)
$deliveryModeint2НетРежим доставки: 1 — непостоянный, 2 — постоянный
// Строка
RabbitMQManager::getInstance()->publish('logs', 'Ошибка в модуле X');

// Массив
RabbitMQManager::getInstance()->publish('orders', [
    'order_id' => 123,
    'user_id'  => 456,
    'total'    => 2500,
]);

publishDelayed()

public function publishDelayed(
    string $queue,
    $message,
    int $delayMs = 5000
): bool

Отправить сообщение с задержкой через dead-letter exchange.

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаЦелевая очередь
$messagestring|array—ДаСообщение
$delayMsint5000НетЗадержка в миллисекундах (5 секунд)
// Отправить напоминание через 10 минут
RabbitMQManager::getInstance()->publishDelayed('reminders', [
    'user_id' => 123,
    'text'    => 'Не забудьте оплатить заказ',
], 600000);

Получение сообщений

consume()

public function consume(
    string $queue,
    callable $callback,
    bool $autoAck = false
): void

Получать сообщения из очереди с обработкой через callback (блокирующий вызов).

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаИмя очереди
$callbackcallable(array): bool|null—ДаФункция-обработчик
$autoAckboolfalseНетАвтоматическое подтверждение

Callback получает декодированное JSON-тело сообщения. Если возвращает false и $autoAck = false, сообщение не подтверждается (requeue).

RabbitMQManager::getInstance()->consume('orders', function (array $data): bool {
    echo "Получен заказ #{$data['order_id']}\n";

    // Обработка заказа
    processOrder($data);

    return true;  // подтверждаем обработку
});

get()

public function get(
    string $queue,
    bool $autoAck = true
): ?array

Получить одно сообщение из очереди (non-blocking).

$message = RabbitMQManager::getInstance()->get('notifications');
if ($message !== null) {
    echo $message['text'];
}

RPC (Remote Procedure Call)

rpcClient()

public function rpcClient(
    string $queue,
    $message,
    int $timeout = 30
): ?array

Выполнить RPC-запрос и получить ответ.

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаОчередь RPC-сервера
$messagestring|array—ДаЗапрос
$timeoutint30НетТаймаут ожидания ответа в секундах
$result = RabbitMQManager::getInstance()->rpcClient(
    'math_queue',
    ['operation' => 'add', 'a' => 2, 'b' => 3],
    10
);

if ($result !== null) {
    echo "Результат: {$result['result']}\n";  // 5
}

rpcServer()

public function rpcServer(
    string $queue,
    callable $callback
): void

Запустить RPC-сервер (блокирующий вызов).

Callback получает декодированное JSON-тело запроса и возвращает результат.

RabbitMQManager::getInstance()->rpcServer('math_queue', function (array $request): array {
    return ['result' => $request['a'] + $request['b']];
});

Управление

close()

public function close(): void

Закрыть канал и соединение.

RabbitMQManager::getInstance()->close();

Примеры

Простая очередь

use IO\Queue\RabbitMQManager;

// Инициализация
RabbitMQManager::init([
    'host'     => 'localhost',
    'user'     => 'guest',
    'password' => 'guest',
]);

// Producer
RabbitMQManager::getInstance()->publish('email_queue', [
    'to'      => 'user@example.com',
    'subject' => 'Привет!',
    'body'    => 'Текст письма',
]);

// Consumer (отдельный процесс)
RabbitMQManager::getInstance()->consume('email_queue', function (array $data): bool {
    sendEmail($data['to'], $data['subject'], $data['body']);
    return true;
});

Очередь с задержкой

// Отправить задачу с задержкой 30 секунд
RabbitMQManager::getInstance()->publishDelayed('cleanup', [
    'type' => 'old_sessions',
], 30000);

// Обработчик
RabbitMQManager::getInstance()->consume('cleanup', function (array $data): bool {
    cleanupOldSessions();
    return true;
});

RPC вызовы

// Сервер (отдельный процесс)
RabbitMQManager::getInstance()->rpcServer('calculator', function (array $request): array {
    return match ($request['op']) {
        'add' => ['result' => $request['a'] + $request['b']],
        'sub' => ['result' => $request['a'] - $request['b']],
        'mul' => ['result' => $request['a'] * $request['b']],
        default => ['error' => 'unknown operation'],
    };
});

// Клиент
$result = RabbitMQManager::getInstance()->rpcClient('calculator', [
    'op' => 'add',
    'a'  => 10,
    'b'  => 20,
]);
echo $result['result'];  // 30

Множественные очереди

$mq = RabbitMQManager::getInstance();

// Обработка разных типов сообщений
$queues = ['orders', 'notifications', 'logs'];

foreach ($queues as $queue) {
    // fork или spawn для каждой очереди
    if (pcntl_fork() === 0) {
        $mq->consume($queue, function (array $data) use ($queue): bool {
            processMessage($queue, $data);
            return true;
        });
        exit;
    }
}

Очередь с retry

$mq = RabbitMQManager::getInstance();

$mq->consume('tasks', function (array $data): bool {
    try {
        processTask($data);
        return true;
    } catch (\Exception $e) {
        // При ошибке — возвращаем в очередь с задержкой
        $retryCount = $data['retry'] ?? 0;
        if ($retryCount < 3) {
            $data['retry'] = $retryCount + 1;
            RabbitMQManager::getInstance()->publishDelayed('tasks', $data, 5000 * $retryCount);
        } else {
            // Отправить в dead letter queue
            RabbitMQManager::getInstance()->publish('tasks_failed', $data);
        }
        return false;
    }
});

Рекомендации

  1. Используйте deliveryMode => 2 для важных сообщений (.persistent)
  2. Обрабатывайте ошибки в consumer — возвращайте false для requeue
  3. Настройте prefetch для балансировки нагрузки
  4. Используйте dead letter queue для необработанных сообщений
  5. Не храните пароли в коде — используйте .env
  6. Закрывайте соединения через close() при завершении

FAQ

Как проверить подключение?

$mq = RabbitMQManager::getInstance();
// Если getInstance() не выбросил исключение — подключение установлено

Как настроить.dead letter queue?

// Очередь с dead letter exchange
$channel = $mq->getClient()->channel();
$channel->queue_declare('tasks', false, true, false, false, false, false, [
    'x-dead-letter-exchange'    => ['S', ''],
    'x-dead-letter-routing-key' => ['S', 'tasks_failed'],
]);

Как ограничить потребление сообщений?

Используйте basic_qos через прямой доступ к каналу:

$mq = RabbitMQManager::getInstance();
$channel = $mq->getClient()->channel();
$channel->basic_qos(null, 10, null);  // prefetch = 10

Что такое RPC в RabbitMQ?

RPC (Remote Procedure Call) — паттерн, при котором клиент отправляет запрос и блочно ожидает ответ. Сервер обрабатывает запрос и отправляет результат обратно во временную очередь.

Как перезапустить consumer?

// Consumer с auto-restart
while (true) {
    try {
        $mq->consume('queue', function (array $data): bool {
            process($data);
            return true;
        });
    } catch (\Exception $e) {
        error_log("Consumer error: " . $e->getMessage());
        sleep(5);
    }
}