diff --git a/src/TrackCenter.Api/appsettings.Development.json b/src/TrackCenter.Api/appsettings.Development.json index 714e902b..a6e86ace 100644 --- a/src/TrackCenter.Api/appsettings.Development.json +++ b/src/TrackCenter.Api/appsettings.Development.json @@ -1,24 +1,8 @@ { - "Serilog": { - "MinimumLevel": { + "Logging": { + "LogLevel": { "Default": "Debug", - "Override": { - "Microsoft.AspNetCore": "Information" - } - }, - "Enrich": [ - "FromLogContext", - "WithMachineName", - "WithProcessId", - "WithThreadId" - ], - "WriteTo": [ - { - "Name": "Console" - } - ] - }, - "itfx": { - "BaseAddress": "http://43.139.210.35:8888/" + "Microsoft.AspNetCore": "Warning" + } } } diff --git a/src/TrackCenter.Api/appsettings.json b/src/TrackCenter.Api/appsettings.json index c41ba880..87513698 100644 --- a/src/TrackCenter.Api/appsettings.json +++ b/src/TrackCenter.Api/appsettings.json @@ -6,6 +6,9 @@ "Microsoft.AspNetCore": "Debug" } }, + "Properties": { + "ServiceName": "track-api" + }, "Enrich": [ "FromLogContext", "WithMachineName", @@ -14,7 +17,11 @@ ], "WriteTo": [ { - "Name": "Console" + "Name": "Console", + "Args": { + "outputTemplate": "[{Timestamp:HH:mm:ss} {Level:u3}] {SourceContext} - {Message:lj}{NewLine}{Exception}", + "formatter": "Serilog.Formatting.Json.JsonFormatter, Serilog" + } } ] }, @@ -27,6 +34,9 @@ "http://localhost:8889" ] }, + "itfx": { + "BaseAddress": "http://43.139.210.35:8888/" + }, "TrackApp": { "TrackRefreshFetchCount": 4, "OrderPullStatusScanInterval": 60, @@ -55,7 +65,8 @@ "HostName": "129.204.22.146", "Port": 5672, "UserName": "rabbit", - "Password": "itd!@#123" + "Password": "itd!@#123", + "DispatchConsumersAsync": true } } }, @@ -67,7 +78,7 @@ "IndexPrefix": "" }, "Kafka": { - "BootstrapServers": "81.71.47.150:9192,81.71.47.150:9292,81.71.47.150:9392", + "BootstrapServers": "81.71.47.150:9092", "Consumer": { // 会话超时时间ms,超时时间未收到客户端心跳，则会移除该客户端同时触发自平衡 "SessionTimeoutMs": 45000, diff --git a/src/TrackCenter.Application/Rabbit/IRabbitHandler.cs b/src/TrackCenter.Application/Rabbit/IRabbitHandler.cs index fb630619..19635f18 100644 --- a/src/TrackCenter.Application/Rabbit/IRabbitHandler.cs +++ b/src/TrackCenter.Application/Rabbit/IRabbitHandler.cs @@ -4,9 +4,7 @@ namespace TrackCenter.Application.Rabbit; public interface IRabbitHandler { - ConsumeOptions Options(); + bool Enable(ConsumeOptions options); - bool Enable(ConsumeOptions options) => options.Queue == Options().Queue; - - Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options); + Task Handle(BasicDeliverEventArgs args); } diff --git a/src/TrackCenter.Application/Rabbit/RabbitClient.cs b/src/TrackCenter.Application/Rabbit/RabbitClient.cs index 9381d5e2..ac0ebdae 100644 --- a/src/TrackCenter.Application/Rabbit/RabbitClient.cs +++ b/src/TrackCenter.Application/Rabbit/RabbitClient.cs @@ -29,16 +29,30 @@ public class RabbitClient( } var consumer = new AsyncEventingBasicConsumer(channel); - consumer.Received += async (model, ea) => + consumer.Received += (model, ea) => { - await using var scope = serviceScopeFactory.CreateAsyncScope(); - foreach (var handler in scope.ServiceProvider.GetServices()) + _ = Task.Run(async () => { - if (handler.Enable(consumeOptions)) + try { - await handler.Handle(scope.ServiceProvider, ea, consumeOptions); + await using var scope = serviceScopeFactory.CreateAsyncScope(); + foreach (var handler in scope.ServiceProvider.GetServices()) + { + if (handler.Enable(consumeOptions)) + { + await handler.Handle(ea); + } + } + channel.BasicAck(ea.DeliveryTag, false); } - } + catch (Exception ex) + { + logger.LogError($"rabbit on queue({consumeOptions.Queue}) received error: {ex}"); + channel.BasicNack(ea.DeliveryTag, false, consumeOptions.FailedRequeue); + } + }); + + return Task.CompletedTask; }; channel.CallbackException += (sender, args) => diff --git a/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs b/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs index 4fea32a3..19eebd82 100644 --- a/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs +++ b/src/TrackCenter.Application/Rabbit/TrackPushCompletedEventHandler.cs @@ -1,4 +1,5 @@ using System.Text; +using Fake.UnitOfWork; using Newtonsoft.Json; using RabbitMQ.Client.Events; using TrackCenter.Application.Dtos.TrackPush; @@ -10,6 +11,7 @@ using TrackCenter.Domain.Aggregates.TrackPushAggregate; namespace TrackCenter.Application.Rabbit; +[ExposeServices(typeof(IRabbitHandler))] public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandler { private readonly IOrderRepository _orderRepository; @@ -27,15 +29,13 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl _returnWarningHandler = returnWarningHandler; } - public ConsumeOptions Options() + public bool Enable(ConsumeOptions options) { - return new ConsumeOptions - { - Queue = GlobalConst.TrackPushCompletedSubscribe - }; + return options.Queue.StartsWith(GlobalConst.TrackPushCompletedSubscribe); } - public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) + [UnitOfWork] + public virtual async Task Handle(BasicDeliverEventArgs args) { var msg = string.Empty; try @@ -45,7 +45,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl if (@event == null) { LogHelper.Warn( - $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); + $"rabbit on queue received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); return; } diff --git a/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs b/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs index 4509c758..aa7196ef 100644 --- a/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs +++ b/src/TrackCenter.Application/Rabbit/TrackRefreshEventHandler.cs @@ -1,6 +1,7 @@ using System.Diagnostics; using System.Text; using DotNetCore.CAP; +using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Logging; using Newtonsoft.Json; @@ -15,6 +16,7 @@ using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; namespace TrackCenter.Application.Rabbit; +[ExposeServices(typeof(IRabbitHandler))] public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler { private readonly ICapPublisher _capPublisher; @@ -35,15 +37,13 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler _logger = logger; } - public ConsumeOptions Options() + public bool Enable(ConsumeOptions options) { - return new ConsumeOptions - { - Queue = GlobalConst.TrackRefreshSubscribe, - }; + return options.Queue.StartsWith(GlobalConst.TrackRefreshSubscribe); } - public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) + [UnitOfWork] + public virtual async Task Handle(BasicDeliverEventArgs args) { var msg = string.Empty; try @@ -53,7 +53,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler if (@event == null) { _logger.LogWarning( - $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); + $"rabbit on queue received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); return; } @@ -226,23 +226,23 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler // 轨迹发生ISO时间 var trackIsoTime = paramConfig.TrackIsoTimeName == null ? null - : trackToken.SelectToken(paramConfig.TrackIsoTimeName)?.Value(); + : trackToken.SelectToken(paramConfig.TrackIsoTimeName)?.Value(); // 轨迹发生UTC时间 var trackUtcTime = paramConfig.TrackUtcTimeName == null ? null - : trackToken.SelectToken(paramConfig.TrackUtcTimeName)?.Value(); + : trackToken.SelectToken(paramConfig.TrackUtcTimeName)?.Value(); // 轨迹发生时间 var trackTime = trackIsoTime?.DateTime ?? (paramConfig.TrackTimeName == null ? null - : trackToken.SelectToken(paramConfig.TrackTimeName)?.Value()); + : trackToken.SelectToken(paramConfig.TrackTimeName)?.Value()); // 轨迹发生时区 var trackTimeZone = (float?) trackIsoTime?.Offset.TotalHours ?? (paramConfig.TrackTimeZoneName == null ? null - : trackToken.SelectToken(paramConfig.TrackTimeZoneName)?.Value()); + : trackToken.SelectToken(paramConfig.TrackTimeZoneName)?.Value()); // 轨迹发生地 var trackPlace = paramConfig.TrackPlaceName == null ? null - : trackToken.SelectToken(paramConfig.TrackPlaceName)?.Value(); + : trackToken.SelectToken(paramConfig.TrackPlaceName)?.Value(); // 轨迹扩展信息，根据输入的多个字段提取报文内容，构建一个新的对象的json形式 var trackExtra = paramConfig.TrackExtraNames == null ? null