Обзор очередей — IO7 Framework

Внимание

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

Обзор

IO7 предоставляет интеграцию с RabbitMQ для организации асинхронной обработки задач. Менеджер очередей поддерживает публикацию/потребление сообщений, отложенные сообщения, RPC-запросы иdead-letter exchange.

Архитектура

IO\Queue\RabbitMQManager   — Менеджер RabbitMQ (singleton)
  ├── publish()             — Публикация сообщений
  ├── publishDelayed()      — Отложенные сообщения (dead-letter exchange)
  ├── consume()             — Блокирующее потребление
  ├── get()                 — Non-blocking получение
  ├── rpcClient()           — RPC-клиент
  ├── rpcServer()           — RPC-сервер
  └── close()               — Закрытие соединения

Зависимости

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

Конфигурация (.env)

RABBITMQ_HOST=localhost
RABBITMQ_PORT=5672
RABBITMQ_USER=guest
RABBITMQ_PASSWORD=guest
RABBITMQ_VHOST=/

Быстрый старт

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

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

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

\RabbitMQManager::getInstance()->publish('orders', [
    'order_id' => 123,
    'status' => 'new',
]);

Потребление сообщений

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

Non-blocking получение

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

Отложенное сообщение

\RabbitMQManager::getInstance()->publishDelayed('tasks', [
    'job' => 'cleanup'
], 10000);

RPC-запрос

$result = \RabbitMQManager::getInstance()->rpcClient('math_queue', [
    'operation' => 'add',
    'a' => 2,
    'b' => 3,
], 10);

Сценарии использования

Очередь задач

// Producer
\RabbitMQManager::getInstance()->publish('email_queue', [
    'to' => 'user@example.com',
    'subject' => 'Welcome',
    'body' => 'Hello!',
]);

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

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

\RabbitMQManager::getInstance()->publishDelayed('reminders', [
    'user_id' => 123,
    'message' => 'Не забудьте оплатить',
], 3600000);

RPC для синхронных вызовов

// Сервер (отдельный процесс)
\RabbitMQManager::getInstance()->rpcServer('math_queue', function (array $request): array {
    return ['result' => $request['a'] + $request['b']];
});

// Клиент
$result = \RabbitMQManager::getInstance()->rpcClient('math_queue', [
    'operation' => 'add',
    'a' => 2,
    'b' => 3,
]);

CLI команды

# Запуск воркера очереди
php iocli queue:work

# Прослушивание очереди
php iocli queue:listen

# Retry failed jobs
php iocli queue:retry

# Show failed jobs
php iocli queue:failed

# Restart workers
php iocli queue:restart

См. также

RabbitMQ