mikhail/parr_api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(nextRunWorker): добавлен еще один backgroundWorker, который смотрит очередь, и пересчитывает nextRun для jobGroupId

Browse Source
This commit is contained in:
Mikhail Trubnikov
2026-02-27 14:05:25 +10:00
parent f7a2170454
commit 10bf68709b
16 changed files with 319 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
PARR.BLL/Domain/Mq/NextRunUpdateMq.cs Normal file
View File

@@ -0,0 +1,10 @@
namespace PARR.BLL.Domain.Mq
{
/// <summary>
/// Запрос на обновление всех nextRun для JobGroup
/// </summary>
public class NextRunUpdateMq
{
public Guid JobGroupId { get; set; }
}
}

1
PARR.EsppTemplateSync/TemplateMQSyncer.cs
View File

@@ -40,7 +40,6 @@ namespace PARR.EsppTemplateSync
public async Task StartAsync()
{
var isConnected = await mqService.InitConsumerAsync(globalSettings!.MqSettings!, SyncTemplateAsync);
if (!isConnected)

2
PARR.NextRun/INextRunManager.cs → PARR.NextRun/INextRunIntervalService.cs
View File

@@ -1,6 +1,6 @@
namespace PARR.NextRun
{
public interface INextRunManager
public interface INextRunIntervalService
{
Task StartAsync();
}

8
PARR.NextRun/INextRunRabbitService.cs Normal file
View File

@@ -0,0 +1,8 @@
namespace PARR.NextRun
{
public interface INextRunRabbitService
{
Task StartAsync();
Task StopAsync();
}
}

3
PARR.NextRun/NextRunInstaller.cs
View File

@@ -17,7 +17,8 @@ namespace PARR.NextRun
configuration.GetSection(nameof(WorkerSettings)).Bind(settings);
services.AddSingleton(settings);
services.AddTransient<INextRunManager, NextRunManager>();
services.AddTransient<INextRunIntervalService, NextRunIntervalService>();
services.AddTransient<INextRunRabbitService, NextRunRabbitService>();
}

11
PARR.NextRun/NextRunManager.cs → PARR.NextRun/NextRunIntervalService.cs
View File

@@ -11,16 +11,19 @@ using PARR.NextRun.Settings;
namespace PARR.NextRun
{
internal class NextRunManager : INextRunManager
/// <summary>
/// Расчет nextRun по интервалу
/// </summary>
internal class NextRunIntervalService : INextRunIntervalService
{
private readonly WorkerSettings workerSettings;
private readonly ILogger<NextRunManager> logger;
private readonly ILogger<NextRunIntervalService> logger;
private readonly IIntervalService intervalService;
private readonly IServiceProvider serviceProvider;
public NextRunManager(
public NextRunIntervalService(
WorkerSettings workerSettings,
ILogger<NextRunManager> logger,
ILogger<NextRunIntervalService> logger,
IIntervalService intervalService,
IServiceProvider serviceProvider
)

163
PARR.NextRun/NextRunRabbitService.cs Normal file
View File

@@ -0,0 +1,163 @@
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using PARR.BLL.Domain.Mq;
using PARR.BLL.Services.Interfaces;
using PARR.Common.Domain;
using PARR.Constants;
using PARR.DAL.NextRunServices;
using PARR.DAL.Services.Interfaces;
using PARR.DAL.Services.Interfaces.Job;
using PARR.NextRun.Settings;
namespace PARR.NextRun
{
/// <summary>
/// Расчет nextRun по заданию из очереди
/// </summary>
internal class NextRunRabbitService : INextRunRabbitService
{
private readonly ILogger<NextRunRabbitService> logger;
private readonly IMqService mqService;
private readonly WorkerSettings workerSettings;
private readonly ITransformService transformService;
private readonly IServiceProvider serviceProvider;
public NextRunRabbitService(
ILogger<NextRunRabbitService> logger,
IMqService mqService,
WorkerSettings workerSettings,
ITransformService transformService,
IServiceProvider serviceProvider
)
{
this.logger = logger;
this.mqService = mqService;
this.workerSettings = workerSettings;
this.transformService = transformService;
this.serviceProvider = serviceProvider;
}
public async Task StartAsync()
{
var isConnected = await mqService.InitConsumerAsync(workerSettings.MqSettings!, HandlerAsync);
if (!isConnected)
throw new Exception("Ошибка при подключении к RabbitMq");
logger.LogInformation("Запущена проверка очереди {QueueName}.", workerSettings.MqSettings!.QueueName);
}
public async Task StopAsync()
{
await mqService.DisposeAsync();
logger.LogInformation("Соединение с очередью {QueueName} закрыто", workerSettings.MqSettings!.QueueName);
}
private async Task HandlerAsync(string msg)
{
logger.LogInformation("Получили запрос: {message}", msg);
var query = transformService.GetModelFromJson<NextRunUpdateMq>(msg);
if (query == null)
return;
if (!await IsValidJobGroupAsync(query.JobGroupId))
{
logger.LogError("Не найдена группа работ с id: {jobGroupId}", query.JobGroupId);
return;
}
await UpdateNextRunAsync(query.JobGroupId);
}
/// <summary>
/// Существует ли JobGroup с таким Id?
/// </summary>
/// <param name="jobGroupId"></param>
/// <returns></returns>
private async Task<bool> IsValidJobGroupAsync(Guid jobGroupId)
{
using (var scope = serviceProvider.CreateScope())
{
var jobGroupService = scope.ServiceProvider.GetRequiredService<IJobGroupService>();
return await jobGroupService.Get().AnyAsync(t => t.Id == jobGroupId);
}
}
/// <summary>
/// Обновить все NextRun в JobGroup
/// </summary>
/// <param name="jobGroupId"></param>
/// <returns></returns>
private async Task UpdateNextRunAsync(Guid jobGroupId)
{
using var scope = serviceProvider.CreateScope();
var templateService = scope.ServiceProvider.GetRequiredService<ITemplateService>();
var nextRunService = scope.ServiceProvider.GetRequiredService<INextRunServiceV2>();
// Берем шаблоны только в статусе Used
var templates = await templateService.Get()
.Where(t =>
t.Job!.GroupId == jobGroupId
&& t.StatusTypeId == TemplateStatusTypeEnum.Used
).ToListAsync();
logger.LogInformation("Найдено шаблонов {count} шт. в статусе Used в группе работ {jobGroupId}", templates.Count, jobGroupId);
if (!templates.Any())
return;
var updatedTemplates = 0;
foreach (var template in templates)
{
var newNextRun = await nextRunService.GetNextRunForTemplateAsync(template.Id, false);
if (newNextRun == null)
{
logger.LogError("При расчете nextRun для шаблона {templateId}, {templateName} вернулся null", template.Id, template.Name);
continue;
}
if (template.NextRun != newNextRun)
{
logger.LogInformation("Обновлен nextRun для шаблона {templateId}, {templateName}, newNextRun: {newNextRun}, oldNextRun: {oldNextRun}",
template.Id, template.Name, newNextRun, template.NextRun);
template.LastRun = template.NextRun;
template.NextRun = newNextRun.Value;
updatedTemplates++;
}
else
{
logger.LogDebug("Не требуется обновлять nextRun для шаблона {templateId}, {templateName}. Рассчитанный и исходный равны. NextRun: {NextRun}",
template.Id, template.Name, template.NextRun);
}
}
if (updatedTemplates > 0)
{
if (await templateService.CommitAsync(new HistoryInitiator { InitiatorComment = "Запрос из очереди на перерасчет всех nextRun для jobGroupId", InitiatorParrComponentId = ParrComponentsEnum.NextRun }))
{
logger.LogInformation("Успешно обновлены nextRun у {count} шаблонов, группа работ: {jobGroupId}", updatedTemplates, jobGroupId);
}
else
{
logger.LogError("При сохранении nextRun для шаблонов {count} шт, произошла ошибка при сохранении в БД. JobGroupId: {jobGroupId}", updatedTemplates, jobGroupId);
}
}
else
{
logger.LogInformation("Для группы работ {jobGroupId}, все nextRun актуальны. Нечего обновлять.", jobGroupId);
}
}
}
}

16
PARR.NextRun/Settings/WorkerSettings.cs
View File

@@ -1,7 +1,21 @@
namespace PARR.NextRun.Settings
using PARR.BLL.Contracts.Interfaces;
namespace PARR.NextRun.Settings
{
internal class WorkerSettings
{
public TimeSpan RepeatEvery { get; set; }
public MqSettings? MqSettings { get; set; }
}
internal class MqSettings : IMqSettings
{
public string HostName { get; set; } = string.Empty;
public string QueueName { get; set; } = string.Empty;
public string User { get; set; } = string.Empty;
public string Password { get; set; } = string.Empty;
public ushort? PrefetchCount { get; set; } = 0;
}
}

22
PARR.NextRunWorker/IntervalWorker.cs Normal file
View File

@@ -0,0 +1,22 @@
using PARR.NextRun;
namespace PARR.NextRunWorker
{
public class IntervalWorker : BackgroundService
{
private readonly INextRunIntervalService nextRunIntervalService;
private readonly ILogger<IntervalWorker> logger;
public IntervalWorker(INextRunIntervalService nextRunIntervalService, ILogger<IntervalWorker> logger)
{
this.nextRunIntervalService = nextRunIntervalService;
this.logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
logger.LogInformation("Запуск {workerName}", nameof(IntervalWorker));
await nextRunIntervalService.StartAsync();
}
}
}

20
PARR.NextRunWorker/Program.cs
View File

@@ -25,7 +25,25 @@ builder.Services.InstallNextRunServices(builder.Configuration);
builder.Configuration.AddNextRunConfigurations(builder.Services);
builder.Services.AddNextRunSettings(builder.Configuration);
builder.Services.AddHostedService<Worker>();
// Задается через переменные окружеия
var runRabbit = builder.Configuration.GetValue<bool>("ENABLE_RABBIT", true);
var runInterval = builder.Configuration.GetValue<bool>("ENABLE_INTERVAL", true);
if (runRabbit)
{
builder.Services.AddHostedService<RabbitWorker>();
}
if (runInterval)
{
builder.Services.AddHostedService<IntervalWorker>();
}
// Настройка краша при ошибке. Если хоть один из воркеров упадет, то падает целиком проект
builder.Services.Configure<HostOptions>(options =>
{
options.BackgroundServiceExceptionBehavior = BackgroundServiceExceptionBehavior.StopHost;
});
var host = builder.Build();
host.Run();

28
PARR.NextRunWorker/RabbitWorker.cs Normal file
View File

@@ -0,0 +1,28 @@
using PARR.NextRun;
namespace PARR.NextRunWorker
{
public class RabbitWorker : BackgroundService
{
private readonly INextRunRabbitService nextRunRabbitService;
private readonly ILogger<RabbitWorker> logger;
public RabbitWorker(INextRunRabbitService nextRunRabbitService, ILogger<RabbitWorker> logger)
{
this.nextRunRabbitService = nextRunRabbitService;
this.logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
logger.LogInformation("Запуск {workerName}", nameof(RabbitWorker));
await nextRunRabbitService.StopAsync();
}
public override Task StopAsync(CancellationToken cancellationToken)
{
nextRunRabbitService.StopAsync().Wait();
return base.StopAsync(cancellationToken);
}
}
}

19
PARR.NextRunWorker/Worker.cs
View File

@@ -1,19 +0,0 @@
using PARR.NextRun;
namespace PARR.NextRunWorker
{
public class Worker : BackgroundService
{
private readonly INextRunManager nextRunManager;
public Worker(INextRunManager nextRunManager)
{
this.nextRunManager = nextRunManager;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await nextRunManager.StartAsync();
}
}
}

5
PARR.NextRunWorker/appsettings.Development.json
View File

@@ -25,5 +25,10 @@
}
}
]
},
"WorkerSettings": {
"MqSettings": {
"HostName": "10.99.253.216"
}
}
}

9
PARR.NextRunWorker/appsettings.json
View File

@@ -23,6 +23,13 @@
}
},
"WorkerSettings": {
"RepeatEvery": "0:50:00"
"RepeatEvery": "0:50:00",
"MqSettings": {
"HostName": "parr-rabbitmq",
"QueueName": "parr-next-run-updater",
"User": "next_run_updater_reader",
"Password": "LJFgsifgFidfsdvi123#!23",
"PrefetchCount": 100
}
}
}

4
PARR.Test/NextRun/NextRunTest.cs
View File

@@ -37,6 +37,10 @@ namespace PARR.Test.NextRun
public async Task Test()
{
//var nextRun = await nextRunServiceV2.GetNextRunForTemplateAsync(Guid.Parse("0bef9892-1672-40b7-a2bd-528f9bd1ef22"), false);
#region тестирование INextRunServiceV2
//рапсределить

27
docker-compose.next-run.yml
View File

@@ -2,11 +2,13 @@ version: '3.4'
# NEXT RUN
services:
parr-next-run:
parr-next-run-interval:
image: harbor.dvgd.rzd/parr/parr-next-run:${tag:-latest}
environment:
- ASPNETCORE_ENVIRONMENT=Production
- TZ=Europe/Moscow
- ENABLE_RABBIT=false
- ENABLE_INTERVAL=true
logging:
driver: fluentd
options:
@@ -15,7 +17,28 @@ services:
fluentd-max-retries: '30'
fluentd-async: 'true'
fluentd-buffer-limit: '52428800'
tag: parr.next-run.serilog
tag: parr.next-run-interval.serilog
deploy:
replicas: 1
networks:
- parr-network
parr-next-run-rabbit:
image: harbor.dvgd.rzd/parr/parr-next-run:${tag:-latest}
environment:
- ASPNETCORE_ENVIRONMENT=Production
- TZ=Europe/Moscow
- ENABLE_RABBIT=true
- ENABLE_INTERVAL=false
logging:
driver: fluentd
options:
fluentd-address: dvgd-efk-01.dvgd.oao.rzd:24224
fluentd-retry-wait: '10s'
fluentd-max-retries: '30'
fluentd-async: 'true'
fluentd-buffer-limit: '52428800'
tag: parr.next-run-rabbit.serilog
deploy:
replicas: 1
networks: