RabbitMQ — IO7 Framework

Внимание

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

Обзор

IO\Queue\RabbitMQManager — менеджер для работы с RabbitMQ. Управляет подключениями, отправкой и получением сообщений. Реализован как singleton через static $instance.

Зависимость

{
    "php-amqplib/php-amqplib": "^3.7"
}

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

Через init()

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

Параметры

ПараметрТипЗначение по умолчаниюОбязателенОписание
hoststring—ДаХост RabbitMQ
portint5672НетПорт
userstring—ДаПользователь
passwordstring—ДаПароль
vhoststring'/'НетVirtual host

Методы

init(array $config): self

Инициализирует синглтон с заданной конфигурацией. При повторном вызове возвращает уже созданный экземпляр.

ПараметрТипЗначение по умолчаниюОбязателенОписание
$configarray<string, mixed>—ДаМассив конфигурации подключения
ВозвратselfЭкземпляр менеджера
$manager = RabbitMQManager::init([
    'host' => 'localhost',
    'user' => 'guest',
    'password' => 'guest',
]);

getInstance(): self

Возвращает текущий экземпляр синглтона.

ПараметрТипЗначение по умолчаниюОбязателенОписание
ВозвратselfЭкземпляр менеджера
Исключение\ExceptionЕсли менеджер не был инициализирован
$manager = RabbitMQManager::getInstance();
$manager->publish('test_queue', 'hello');

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

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

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаИмя очереди
$messagestring|array—ДаТекстовое сообщение или массив (сериализуется в JSON)
$exchangestring''НетИмя exchange (по умолчанию '')
$deliveryModeint2НетРежим доставки: 1=непостоянный, 2=постоянный (по умолчанию 2)
ВозвратboolВсегда true при успешной отправке
RabbitMQManager::getInstance()->publish('orders', ['order_id' => 123, 'status' => 'new']);
RabbitMQManager::getInstance()->publish('logs', 'error occurred', '', 1);

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

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

Создаёт промежуточную очередь с суффиксом .dlq, которая после истечения TTL перенаправляет сообщение в целевую очередь.

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаИмя целевой очереди
$messagestring|array—ДаСообщение
$delayMsint5000НетЗадержка в миллисекундах (по умолчанию 5000)
ВозвратboolВсегда true
RabbitMQManager::getInstance()->publishDelayed('tasks', ['job' => 'cleanup'], 10000);

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

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

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаИмя очереди
$callbackcallable(array<string, mixed>): bool|null—ДаФункция-обработчик
$autoAckboolfalseНетАвтоматическое подтверждение (по умолчанию false)
RabbitMQManager::getInstance()->consume('orders', function (array $data): bool {
    echo "Получен заказ: " . $data['order_id'];
    return true;
});

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

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

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаИмя очереди
$autoAckbooltrueНетАвтоматическое подтверждение (по умолчанию true)
Возвратarray|nullДекодированное тело сообщения или null
$message = RabbitMQManager::getInstance()->get('notifications');
if ($message !== null) {
    echo $message['text'];
}

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

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

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

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

Запустить RPC-сервер, обрабатывающий запросы из очереди. Блокирующий вызов.

ПараметрТипЗначение по умолчаниюОбязателенОписание
$queuestring—ДаИмя очереди
$callbackcallable(array<string, mixed>): mixed—ДаФункция-обработчик
RabbitMQManager::getInstance()->rpcServer('math_queue', function (array $request): array {
    return ['result' => $request['a'] + $request['b']];
});

close(): void

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

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

Примеры

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

// Producer (веб-приложение)
\RabbitMQManager::init(['host' => 'localhost', 'user' => 'guest', 'password' => 'guest']);
\RabbitMQManager::getInstance()->publish('email_queue', [
    'to' => 'user@example.com',
    'subject' => 'Welcome!',
    'body' => 'Hello world!',
]);

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

Отложенная обработка

// Планирование задачи
\RabbitMQManager::getInstance()->publishDelayed('cleanup', [
    'type' => 'old_sessions',
], 3600000);

// Воркер (отдельный процесс)
\RabbitMQManager::getInstance()->consume('cleanup', function (array $data): bool {
    if ($data['type'] === 'old_sessions') {
        cleanupOldSessions();
    }
    return true;
});

RPC-вызов

// Сервер (отдельный процесс)
\RabbitMQManager::getInstance()->rpcServer('image_resize', function (array $request): array {
    $resized = resizeImage($request['path'], $request['width']);
    return ['resized_path' => $resized];
});

// Клиент (веб-приложение)
$result = \RabbitMQManager::getInstance()->rpcClient('image_resize', [
    'path' => '/uploads/photo.jpg',
    'width' => 800,
], 30);
echo $result['resized_path'];