Приоритизация обработки сообщений в DATAREON Platform: настраиваем очереди по типу данных и свойствам

С вами снова Илья Вязников, инженер сопровождения СОФРОС. Продолжаю делится практическими приёмами и полезными настройками платформы.

Когда в интеграционной платформе одновременно обрабатываются тысячи сообщений разных типов, стандартной FIFO-очереди часто бывает недостаточно. Критически важные сообщения должны обрабатываться в первую очередь, даже если система сильно загружена.

В этой статье разберём, как настроить гибкую систему приоритезации очередей в DATAREON Platform на основе типа данных и дополнительных свойств сообщения.

Задача

Нужно было обеспечить приоритетную обработку сообщений определённых типов, либо с определенными свойствами. Сообщения с высоким приоритетом должны проходить раньше остальных, а система при этом оставалась простой в поддержке и расширении.

Решение

DATAREON Platform позволяет создавать правила обработки очередей, в которых можно:

• Распределять входящие сообщения по разным очередям

• Извлекать сообщения для обработки в заданном порядке приоритетов.

Настройка

1. Создание очередей

Правила обработки создаются в Центре Настройки в разделе Обработка данных - Правила обработки очередей.

В правиле обработки очередей на вкладке «Основные» указываем название правила, и включаем его отметив тик-бокс.

Добавляем необходимые очереди, например:

• Queue_HighPriority - Высокий приоритет

• Queue_NormalPriority - Обычный приоритет

• Queue_LowPriority - Низкий приоритет (по умолчанию)

• Queue_Property1Priority, Queue_Property2Priority - очереди по значению свойства

2. Правило размещения сообщений (Placement Rule)

На вкладке «Правило размещения» настраиваем логику распределения:

// Словарь для определения очереди по типу данных
// Если сообщение содержит определенный тип данных, оно будет отправлено в соответствующую очередь
OwnDictionary<Metadata.DataTypes, QueueState> DictionaryOfValidQueuesByType = new()
{
  [Metadata.DataTypes.Тип_1_18235] = Queues.Queue_NormalPriority,
  [Metadata.DataTypes.Тип_2_18235] = Queues.Queue_HighPriority
};


// Словарь для определения очереди по строковому свойству "stringProperty"
// Если сообщение содержит определенное значение свойства, оно будет отправлено в соответствующую очередь
OwnDictionary<string, QueueState> DictionaryOfValidQueuesByProperty = new()
{
  ["Property_1"] = Queues.Queue_Property1Priority,
  ["Property_2"] = Queues.Queue_Property2Priority
};


// Назначение приоритетных очередей
// 1. Проверяем, есть ли соответствие типа данных в словаре DictionaryOfValidQueuesByType
if (!DictionaryOfValidQueuesByType.TryGetValue(InitMessage.GetDataType(), out QueueState Queue_Local))
{
  // 2. Если нет, проверяем соответствие строкового свойства в словаре DictionaryOfValidQueuesByProperty
  If
  (!DictionaryOfValidQueuesByProperty.TryGetValue(InitMessage.GetStringProperty("stringProperty"), out Queue_Local))
    {
    // 3. Если нет ни по типу данных, ни по строковому свойству – назначаем очередь по умолчанию (LowPriority)
    Queue_Local = Queues.Queue_LowPriority; // Очередь по умолчанию
    }
}

// Присваиваем выбранную очередь
Queue = Queue_Local;

Как работать с этим кодом:

• Добавление новых типов данных – если появится новый тип данных, который нужно направлять в определенную очередь, просто добавьте новую запись в DictionaryOfValidQueuesByType.

• Добавление новых значений stringProperty – если появится новое значение stringProperty, которое должно попадать в конкретную очередь, добавьте его в DictionaryOfValidQueuesByProperty.

• Если ни один параметр не совпадает – сообщение автоматически попадет в очередь Queue_LowPriority.

• Логика приоритета:

  • В первую очередь проверяется соответствие по типу данных.

  • Если тип данных не определяет очередь – проверяется значение свойства stringProperty.

  • Если и это не дало результата – сообщение уходит в очередь по умолчанию.

3. Правило извлечения сообщений (Extraction Rule)

На вкладке «Правило извлечения» определяем порядок обработки:

// Формируем список очередей в порядке их приоритета (от высокой к низкой)
var priorityQueues = new List<QueueState>
{
    Queues.Queue_HighPriority,     // Очередь с наивысшим приоритетом
    Queues.Queue_NormalPriority,   // Очередь со средним приоритетом
    Queues.Queue_Property1Priority, // Очередь, связанная с Property_1
    Queues.Queue_Property2Priority  // Очередь, связанная с Property_2
};


// Находим первую непустую очередь или устанавливаем очередь по умолчанию
Queue = priorityQueues.FirstOrDefault(q => q.Count > 0) ?? Queues.Queue_LowPriority;

Как работает код:

• Определение приоритетных очередей

  • Очереди расположены в порядке приоритета – от самой важной (Queue_HighPriority) до менее значимых (Queue_Property2Priority).

  • Если появятся новые приоритетные очереди, их можно добавить в список priorityQueues в нужном порядке.

• Поиск первой непустой очереди

  • Метод FirstOrDefault(q => q.Count > 0) ищет первую очередь, в которой есть сообщения.

  • Как только находит такую очередь, она выбирается для обработки.

• Если все очереди пусты

  • Оператор ?? означает, что если FirstOrDefault() не нашел не пустую очередь (вернул null), используем Queue_LowPriority как очередь по умолчанию.

Как работать с этим кодом:

• Если нужно изменить приоритет очередей, просто поменяйте порядок элементов в priorityQueues.

• Если требуется добавить новую очередь, добавьте её в priorityQueues в нужное место (в зависимости от её приоритета).

• Очередь Queue_LowPriority остается запасной, если все остальные очереди пусты.

4. Применение правила

Созданное правило обработки очередей указывается в настройках нужных внешних систем либо сервисов в поле Правило обработки очередей.

После применения конфигурации в Центре мониторинга появляются новые очереди, по которым можно отслеживать нагрузку и состояние:

Результат

После внедрения настройки сообщения будут обрабатываться в соответствии с настроенным приоритетом. Критически важные сообщения стабильно проходят раньше остальных, что позволит повысить предсказуемость и надёжность интеграционных процессов.

Преимущества подхода

• Простота расширения - достаточно добавить новую запись в словарь.

• Явный приоритет - порядок очередей в списке priorityQueues полностью определяет стратегию.

• Универсальность - работает как по типу данных, произвольным свойствам сообщения, так и по источнику, обработчику не рассмотренным в примере.

• Удобный мониторинг - каждая очередь видна в Центре мониторинга.

Заключение

Приоритезация очередей - мощный инструмент DATAREON Platform, который помогает решать задачи высоконагруженных интеграций. Подход, описанный выше, легко адаптировать под свои типы сообщений и бизнес-правила.

Подробнее с правилами обработки очередей вы можете ознакомиться в документации:

Правила обработки очередей

Настройка правил с помощью кода C#

Если у вас есть вопросы по реализации или нужны уточнения - пишите в комментариях.

Теги: DATAREON, Platform, очереди, приоритизация, message queue, интеграция, C#, middleware