From a7f7e732f68befefad29ce71bfa6e1021ce147b5 Mon Sep 17 00:00:00 2001 From: Mikhail Trubnikov Date: Mon, 17 Nov 2025 15:36:24 +1000 Subject: [PATCH] =?UTF-8?q?feat(bll):=20=D0=BE=D0=B1=D0=BD=D0=BE=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20MqServiceV2,=20=D0=B4=D0=BE=D0=B1=D0=B0?= =?UTF-8?q?=D0=B2=D0=BB=D0=B5=D0=BD=D0=B0=20=D0=BE=D0=B1=D1=80=D0=B0=D0=B1?= =?UTF-8?q?=D0=BE=D1=82=D0=BA=D0=B0=20=D0=BE=D1=88=D0=B8=D0=B1=D0=BE=D0=BA?= =?UTF-8?q?=20=D0=BF=D1=80=D0=B8=20=D0=B2=D1=8B=D0=BF=D0=BE=D0=BB=D0=BD?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=B4=D0=B5=D0=BB=D0=B5=D0=B3=D0=B0?= =?UTF-8?q?=D1=82=D0=B0.=20=D0=9E=D0=B1=D0=BD=D0=BE=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=20image=20rabbitMq=20=D0=B4=D0=BE=20=D0=B2=D0=B5=D1=80?= =?UTF-8?q?=D1=81=D0=B8=D0=B8=204.2.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Services/Implementations/MqServiceV2.cs | 30 ++++++++++++++++--- PARR.GeneratorTemplatesWorker/Worker.cs | 15 ++++------ docker-compose.espp-template-sync.yml | 1 + 3 files changed, 32 insertions(+), 14 deletions(-) diff --git a/PARR.BLL/Services/Implementations/MqServiceV2.cs b/PARR.BLL/Services/Implementations/MqServiceV2.cs index a299ef13..b37f827d 100644 --- a/PARR.BLL/Services/Implementations/MqServiceV2.cs +++ b/PARR.BLL/Services/Implementations/MqServiceV2.cs @@ -75,12 +75,26 @@ namespace PARR.BLL.Services.Implementations consumer.ReceivedAsync += async (ch, ea) => { - var content = Encoding.UTF8.GetString(ea.Body.ToArray()); + //var content = Encoding.UTF8.GetString(ea.Body.ToArray()); + var content = Encoding.UTF8.GetString(ea.Body.Span); logger.LogDebug($"Получено сообщение: {content}"); - await messageHandler.Invoke(content); + try + { + await messageHandler.Invoke(content); + await consumerChannel.BasicAckAsync(ea.DeliveryTag, false); + } + catch (Exception ex) + { + logger.LogError(ex, $"Ошибка при обработке сообщения: {content}"); - await consumerChannel.BasicAckAsync(ea.DeliveryTag, false); + //requeue: true - сообщение возвращаем в очередь в случае ошибки. Это немного опасно, + // если ошибка в формате сообщения, то она никогда не устранится и вечно будет добавляться/удаляться в очередь + //todo: тут можно реализовать DLQ с кол-вом попыток, после не успешных попыток, перемещать эти сообщения в другую очередь + // сделал false, выбрасывать на свежий воздух куда подальше эти кривые сообщения которые обработал с ошибками + await consumerChannel.BasicNackAsync(ea.DeliveryTag, false, requeue: false); + //await consumerChannel.BasicNackAsync(ea.DeliveryTag, false, requeue: true); + } }; var consumerTag = await consumerChannel.BasicConsumeAsync(mqSettings.QueueName, false, consumer); @@ -124,6 +138,12 @@ namespace PARR.BLL.Services.Implementations //props.Expiration = "60000"; props.ContentType = "text/plain";//"application/json"; + channel.BasicReturnAsync += async (sender, ea) => + { + var body = Encoding.UTF8.GetString(ea.Body.Span); + logger.LogWarning($"Сообщение не было доставлено в очередь: {ea.Exchange} -> {ea.RoutingKey}. Тело: {body}"); + }; + foreach (var msg in msgList) { var body = Encoding.UTF8.GetBytes(msg); @@ -157,11 +177,13 @@ namespace PARR.BLL.Services.Implementations } } + private async Task QueueDeclareAsync(IChannel channel, IMqSettings mqSettings) { //https://www.rabbitmq.com/lazy-queues.html // lazy - ленивой очереди больше нет - var args = new Dictionary { { "x-queue-mode", "lazy" } }; + //var args = new Dictionary { { "x-queue-mode", "lazy" } }; + var args = new Dictionary { }; //Объявляем очередь с которой будем работать. //Если такой очереди ещё нет, то создатся. diff --git a/PARR.GeneratorTemplatesWorker/Worker.cs b/PARR.GeneratorTemplatesWorker/Worker.cs index 18d0c73d..6e6a01c7 100644 --- a/PARR.GeneratorTemplatesWorker/Worker.cs +++ b/PARR.GeneratorTemplatesWorker/Worker.cs @@ -15,19 +15,14 @@ namespace PARR.GeneratorTemplatesWorker protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - await generatorTemplate.StartAsync(); - //while (!stoppingToken.IsCancellationRequested) - //{ - // _logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now); - // await Task.Delay(1000, stoppingToken); - //} + await generatorTemplate.StartAsync(); + await Task.Delay(Timeout.Infinite, stoppingToken); } - public override Task StopAsync(CancellationToken cancellationToken) + public override async Task StopAsync(CancellationToken cancellationToken) { - generatorTemplate.StopAsync().Wait(); - - return base.StopAsync(cancellationToken); + await generatorTemplate.StopAsync(); + await base.StopAsync(cancellationToken); } } } \ No newline at end of file diff --git a/docker-compose.espp-template-sync.yml b/docker-compose.espp-template-sync.yml index f59d1de7..212ca34b 100644 --- a/docker-compose.espp-template-sync.yml +++ b/docker-compose.espp-template-sync.yml @@ -27,6 +27,7 @@ services: parr-rabbitmq: image: rabbitmq:3.12.4-management + #image: rabbitmq:4.2.0-management #container_name: rabbitmq hostname: parr-rabbitmq environment: