From e89e86945febcdfb5b4a693c60aa732db46fdc84 Mon Sep 17 00:00:00 2001 From: Sasha Rybkin Date: Thu, 5 Nov 2026 11:58:07 +0300 Subject: [PATCH] =?UTF-8?q?#131279=20=D0=9D=D0=B5=D0=B2=D0=B5=D1=80=D0=BD?= =?UTF-8?q?=D0=BE=D0=B5=20=D0=BA=D0=BE=D0=BD=D1=84=D0=B8=D0=B3=D1=83=D1=80?= =?UTF-8?q?=D0=B8=D1=80=D0=BE=D0=B2=D0=B0=D0=BD=D0=B8=D0=B5=20=D0=BE=D1=87?= =?UTF-8?q?=D0=B5=D1=80=D0=B5=D0=B4=D0=B5=D0=B9=20=D0=BF=D1=80=D0=B8=20?= =?UTF-8?q?=D0=BF=D0=B5=D1=80=D0=B5=D0=B7=D0=B0=D0=BF=D1=83=D1=81=D0=BA?= =?UTF-8?q?=D0=B5=20=D1=81=D0=B5=D1=80=D0=B2=D0=B8=D1=81=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AmqpHelper.php | 49 ++++++++++++++++++++++++++++++++------ console/AmqpController.php | 4 ---- 2 files changed, 42 insertions(+), 11 deletions(-) diff --git a/AmqpHelper.php b/AmqpHelper.php index 296a6bf..b931e72 100644 --- a/AmqpHelper.php +++ b/AmqpHelper.php @@ -88,6 +88,11 @@ class AmqpHelper public function getQueue($routeKey) { $queue = $this->getQueueConfig($routeKey); + if (isset($queue['queue']) && $queue['queue'] === false) { + $error = 'Конфигурация ' . $routeKey . ' предназначена только для публикации и не имеет физической очереди.'; + AmqpLoger::log('error', $error, 'system'); + throw new \Exception($error); + } $queueName = $queue['name']; if(!isset(self::$arQueue[$queueName])) { @@ -120,13 +125,14 @@ class AmqpHelper public function getQueueConfig($routeKey) { $project = $this->getProject(); - if (!isset($project['queues'][$routeKey]) || empty($project['queues'][$routeKey])) + $queue = $project['queues'][$routeKey] ?? []; + if (empty($queue)) { $error = 'Ключ маршрутизации '.$routeKey.' не найден'; AmqpLoger::log('error', $error, 'system'); throw new \Exception($error); } - elseif(!isset ($project['queues'][$routeKey]["name"], $project['queues'][$routeKey]["routing"], $project['queues'][$routeKey]["flags"], $project['queues'][$routeKey]["heartbeat"])) + elseif(!isset($queue["name"], $queue["routing"], $queue["flags"], $queue["heartbeat"])) { $error = 'Не верный формат '.$routeKey; AmqpLoger::log('error', $error, 'system'); @@ -134,7 +140,7 @@ class AmqpHelper } else { - return $project['queues'][$routeKey]; + return $queue; } } @@ -148,6 +154,14 @@ class AmqpHelper AmqpLoger::log('error', $error, 'system'); throw new \Exception($error); } + foreach ($this->project["queues"] as $key =>$value) + { + if (isset($value['queue']) && $value['queue'] === false) + { + continue; + } + $this->getQueueObject($key); + } } public function getProject() @@ -189,17 +203,38 @@ class AmqpHelper } /** - * Добавляем произвольную запись в очередь + * Добавляем произвольную запись в Exchange по ключам маршрутизации. + * Поддерживает отправку как в один ключ, так и циклом в несколько ключей из конфига. + * * @param string $projectName синоним обменника (используется в коде) - * @param string $routeKey псевдоним очереди - * @param array $params Поля передаваемого объекта + * @param string $routeKey псевдоним очереди / секции в конфиге + * @param array $sendParams Поля передаваемого объекта */ public static function publishRaw($projectName, $routeKey, $sendParams) { $amqp = new AmqpHelper; $amqp->setProject($projectName); $queue = $amqp->getQueueConfig($routeKey); - $amqp->getExchange($queue)->publish(json_encode($sendParams), $queue['name'], AMQP_NOPARAM, ['delivery_mode' => 2, 'content_type' => 'application/json']); + + // Получаем массив ключей маршрутизации из конфига. + // Если массив пустой или некорректный, откатываемся к дефолтному названию очереди. + $routingKeys = (!empty($queue['routing']) && is_array($queue['routing'])) + ? $queue['routing'] + : [$queue['name']]; + + $exchange = $amqp->getExchange($queue); + $payload = json_encode($sendParams); + + // Отправляем сообщение для каждого ключа маршрутизации + foreach ($routingKeys as $routingKey) { + $exchange->publish( + $payload, + $routingKey, + AMQP_NOPARAM, + ['delivery_mode' => 2, 'content_type' => 'application/json'] + ); + } + AmqpLoger::log('producer', $sendParams, $routeKey); return $sendParams; } diff --git a/console/AmqpController.php b/console/AmqpController.php index dd92f0a..e7c74a4 100644 --- a/console/AmqpController.php +++ b/console/AmqpController.php @@ -32,10 +32,6 @@ class AmqpController extends Controller { $amqp = new AmqpHelper; $amqp->setProject($keyProject); - foreach ($project["queues"] as $key =>$value) - { - $amqp->getQueueObject($key); - } } return ExitCode::OK; }