Событийная шина и кэширование

1. Архитектура событийно-ориентированного взаимодействия

В программном комплексе «UrFU HeatBalance» устранение жёсткой связанности между микросервисами достигнуто за счёт внедрения событийно-ориентированной архитектуры (Event-Driven Architecture, EDA) на базе брокера сообщений RabbitMQ 3.x и фреймворка MassTransit.

Использование асинхронной передачи сообщений гарантирует:

• Непрерывность приёма телеметрии: физический датчик или контроллер получает мгновенное подтверждение приёма пакета (HTTP 202 Accepted), в то время как ресурсоёмкие термодинамические вычисления и запись в базу данных выполняются асинхронно.
• Надёжность доставки (At-Least-Once Delivery): в случае временной недоступности хранилища истории расчётов или вычислительного сервиса сообщения сохраняются в устойчивых (durable) очередях брокера.
• Масштабируемость: консьюмеры очередей могут независимо масштабироваться в нескольких параллельных экземплярах без изменения кода отправителей.

---

2. Контракты событий и команд

Все контракты сообщений вынесены в общую разделяемую сборку HeatBalance.Contracts:

2.1. TelemetryReceivedEvent (Событие приёма телеметрии)

Публикуется сервисом HeatBalance.IoTService в момент успешной валидации входящего пакета от датчика:

public record TelemetryReceivedEvent
{
    public Guid LogId { get; init; }
    public Guid DeviceId { get; init; }
    public string SensorType { get; init; }
    public string CalculationType { get; init; }
    public DateTime Timestamp { get; init; }
    public string PayloadJson { get; init; }
}

2.2. SaveRunCommand (Команда сохранения результатов расчёта)

Отправляется в очередь сервиса HeatBalance.RunService после завершения вычисления теплового баланса:

public record SaveRunCommand
{
    public Guid RunId { get; init; }
    public string CalculationType { get; init; }
    public DateTime Timestamp { get; init; }
    public double Efficiency { get; init; }
    public string InputPayloadJson { get; init; }
    public string ResultPayloadJson { get; init; }
    public bool IsIoT { get; init; }
    public string? OwnerId { get; init; }
}

2.3. DatasetDeletedEvent (Событие удаления датасета)

Публикуется сервисом HeatBalance.DatasetService при удалении шаблона технологического режима для каскадного оповещения связанных IoT-групп:

public record DatasetDeletedEvent
{
    public Guid DatasetId { get; init; }
    public string OwnerId { get; init; }
    public DateTime Timestamp { get; init; }
}

---

3. Топология очередей и обработка сбоев

3.1. Структура очередей в RabbitMQ:

• heatbalance-telemetry-queue: очередь входящих телеметрических событий, обрабатываемая консьюмером TelemetryReceivedConsumer.
• heatbalance-save-run-queue: очередь сохранения расчётных ведомостей, обрабатываемая консьюмером SaveRunConsumer в RunService.
• heatbalance-dataset-events: очередь оповещения об изменениях нормативно-справочных данных.

3.2. Политики повторных попыток (Retry Policies):

Для предотвращения потери сообщений при кратковременных сбоях сети или СУБД в MassTransit настроен механизм повторных попыток:

• Инкрементальный перезапуск: 3 попытки с экспоненциальной задержкой (2, 5, 10 секунд).
• Dead-Letter Exchange (DLX): при невозможности десериализации или фатальной ошибке структуры сообщение перемещается в очередь *-error для последующего анализа администратором.

---

4. Сквозная Sequence-диаграмма обработки телеметрии

На диаграмме представлен полный жизненный цикл пакета от момента замера на датчике до сохранения в историю:

---

5. Распределённое кэширование на базе Redis 7.0

Для повышения быстродействия системы и защиты реляционной СУБД от пиковых нагрузок в архитектуру включён распределённый сервис Redis 7.0.

5.1. Паттерн кэширования (Cache-Aside Pattern)

В микросервисе HeatBalance.DatasetService реализована следующая схема работы:

5.2. Структура ключей и время жизни (TTL):

• dataset:{id}: сериализованное представление структуры параметров датасета. Время жизни: 24 часа (86400 секунд).

5.3. Механизм инвалидации кэша:

Для предотвращения рассинхронизации данных (Stale Data) при выполнении любых мутирующих операций над датасетом:

• При обновлении (PUT /api/v1/datasets/{id}) ключ dataset:{id} удаляется из Redis немедленно после фиксации транзакции в PostgreSQL.
• При удалении (DELETE /api/v1/datasets/{id}) ключ немедленно стирается.

5.4. Результаты внедрения:

• Снижение времени отклика при повторных запросах нормативно-технологических датасетов с 35–50 мс до менее 2 мс.
• Полная изоляция PostgreSQL от паразитных повторяющихся запросов при циклическом опросе интерфейсом.