#131279 Неверное конфигурирование очередей при перезапуске сервиса

This commit is contained in:
srybkin committed 2026-11-05 11:58:07 +03:00
1 parent c9f761bd69
commit e89e86945f
2 files changed
+42 -11

No files matched your search

+42 -7
View File
@@ -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;
}