fix:track refresh

This commit is contained in:
xiaolipro 2026-01-15 15:29:27 +08:00
parent fa53bc160e
commit b353eed8c4
6 changed files with 59 additions and 52 deletions

View File

@ -1,24 +1,8 @@
{ {
"Serilog": { "Logging": {
"MinimumLevel": { "LogLevel": {
"Default": "Debug", "Default": "Debug",
"Override": { "Microsoft.AspNetCore": "Warning"
"Microsoft.AspNetCore": "Information"
} }
},
"Enrich": [
"FromLogContext",
"WithMachineName",
"WithProcessId",
"WithThreadId"
],
"WriteTo": [
{
"Name": "Console"
}
]
},
"itfx": {
"BaseAddress": "http://43.139.210.35:8888/"
} }
} }

View File

@ -6,6 +6,9 @@
"Microsoft.AspNetCore": "Debug" "Microsoft.AspNetCore": "Debug"
} }
}, },
"Properties": {
"ServiceName": "track-api"
},
"Enrich": [ "Enrich": [
"FromLogContext", "FromLogContext",
"WithMachineName", "WithMachineName",
@ -14,7 +17,11 @@
], ],
"WriteTo": [ "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" "http://localhost:8889"
] ]
}, },
"itfx": {
"BaseAddress": "http://43.139.210.35:8888/"
},
"TrackApp": { "TrackApp": {
"TrackRefreshFetchCount": 4, "TrackRefreshFetchCount": 4,
"OrderPullStatusScanInterval": 60, "OrderPullStatusScanInterval": 60,
@ -55,7 +65,8 @@
"HostName": "129.204.22.146", "HostName": "129.204.22.146",
"Port": 5672, "Port": 5672,
"UserName": "rabbit", "UserName": "rabbit",
"Password": "itd!@#123" "Password": "itd!@#123",
"DispatchConsumersAsync": true
} }
} }
}, },
@ -67,7 +78,7 @@
"IndexPrefix": "" "IndexPrefix": ""
}, },
"Kafka": { "Kafka": {
"BootstrapServers": "81.71.47.150:9192,81.71.47.150:9292,81.71.47.150:9392", "BootstrapServers": "81.71.47.150:9092",
"Consumer": { "Consumer": {
// 会话超时时间ms,超时时间未收到客户端心跳,则会移除该客户端同时触发自平衡 // 会话超时时间ms,超时时间未收到客户端心跳,则会移除该客户端同时触发自平衡
"SessionTimeoutMs": 45000, "SessionTimeoutMs": 45000,

View File

@ -4,9 +4,7 @@ namespace TrackCenter.Application.Rabbit;
public interface IRabbitHandler public interface IRabbitHandler
{ {
ConsumeOptions Options(); bool Enable(ConsumeOptions options);
bool Enable(ConsumeOptions options) => options.Queue == Options().Queue; Task Handle(BasicDeliverEventArgs args);
Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options);
} }

View File

@ -29,16 +29,30 @@ public class RabbitClient(
} }
var consumer = new AsyncEventingBasicConsumer(channel); var consumer = new AsyncEventingBasicConsumer(channel);
consumer.Received += async (model, ea) => consumer.Received += (model, ea) =>
{
_ = Task.Run(async () =>
{
try
{ {
await using var scope = serviceScopeFactory.CreateAsyncScope(); await using var scope = serviceScopeFactory.CreateAsyncScope();
foreach (var handler in scope.ServiceProvider.GetServices<IRabbitHandler>()) foreach (var handler in scope.ServiceProvider.GetServices<IRabbitHandler>())
{ {
if (handler.Enable(consumeOptions)) if (handler.Enable(consumeOptions))
{ {
await handler.Handle(scope.ServiceProvider, ea, 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) => channel.CallbackException += (sender, args) =>

View File

@ -1,4 +1,5 @@
using System.Text; using System.Text;
using Fake.UnitOfWork;
using Newtonsoft.Json; using Newtonsoft.Json;
using RabbitMQ.Client.Events; using RabbitMQ.Client.Events;
using TrackCenter.Application.Dtos.TrackPush; using TrackCenter.Application.Dtos.TrackPush;
@ -10,6 +11,7 @@ using TrackCenter.Domain.Aggregates.TrackPushAggregate;
namespace TrackCenter.Application.Rabbit; namespace TrackCenter.Application.Rabbit;
[ExposeServices(typeof(IRabbitHandler))]
public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandler public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandler
{ {
private readonly IOrderRepository _orderRepository; private readonly IOrderRepository _orderRepository;
@ -27,15 +29,13 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
_returnWarningHandler = returnWarningHandler; _returnWarningHandler = returnWarningHandler;
} }
public ConsumeOptions Options() public bool Enable(ConsumeOptions options)
{ {
return new ConsumeOptions return options.Queue.StartsWith(GlobalConst.TrackPushCompletedSubscribe);
{
Queue = GlobalConst.TrackPushCompletedSubscribe
};
} }
public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) [UnitOfWork]
public virtual async Task Handle(BasicDeliverEventArgs args)
{ {
var msg = string.Empty; var msg = string.Empty;
try try
@ -45,7 +45,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
if (@event == null) if (@event == null)
{ {
LogHelper.Warn( 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; return;
} }

View File

@ -1,6 +1,7 @@
using System.Diagnostics; using System.Diagnostics;
using System.Text; using System.Text;
using DotNetCore.CAP; using DotNetCore.CAP;
using Fake.UnitOfWork;
using InterfaceForward.Domain.Shared.Helpers; using InterfaceForward.Domain.Shared.Helpers;
using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging;
using Newtonsoft.Json; using Newtonsoft.Json;
@ -15,6 +16,7 @@ using TrackCenter.Domain.Aggregates.TrackDetailsAggregate;
namespace TrackCenter.Application.Rabbit; namespace TrackCenter.Application.Rabbit;
[ExposeServices(typeof(IRabbitHandler))]
public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
{ {
private readonly ICapPublisher _capPublisher; private readonly ICapPublisher _capPublisher;
@ -35,15 +37,13 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
_logger = logger; _logger = logger;
} }
public ConsumeOptions Options() public bool Enable(ConsumeOptions options)
{ {
return new ConsumeOptions return options.Queue.StartsWith(GlobalConst.TrackRefreshSubscribe);
{
Queue = GlobalConst.TrackRefreshSubscribe,
};
} }
public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) [UnitOfWork]
public virtual async Task Handle(BasicDeliverEventArgs args)
{ {
var msg = string.Empty; var msg = string.Empty;
try try
@ -53,7 +53,7 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
if (@event == null) if (@event == null)
{ {
_logger.LogWarning( _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; return;
} }
@ -226,23 +226,23 @@ public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler
// 轨迹发生ISO时间 // 轨迹发生ISO时间
var trackIsoTime = paramConfig.TrackIsoTimeName == null var trackIsoTime = paramConfig.TrackIsoTimeName == null
? null ? null
: trackToken.SelectToken(paramConfig.TrackIsoTimeName)?.Value<DateTimeOffset>(); : trackToken.SelectToken(paramConfig.TrackIsoTimeName)?.Value<DateTimeOffset?>();
// 轨迹发生UTC时间 // 轨迹发生UTC时间
var trackUtcTime = paramConfig.TrackUtcTimeName == null var trackUtcTime = paramConfig.TrackUtcTimeName == null
? null ? null
: trackToken.SelectToken(paramConfig.TrackUtcTimeName)?.Value<DateTime>(); : trackToken.SelectToken(paramConfig.TrackUtcTimeName)?.Value<DateTime?>();
// 轨迹发生时间 // 轨迹发生时间
var trackTime = trackIsoTime?.DateTime ?? (paramConfig.TrackTimeName == null var trackTime = trackIsoTime?.DateTime ?? (paramConfig.TrackTimeName == null
? null ? null
: trackToken.SelectToken(paramConfig.TrackTimeName)?.Value<DateTime>()); : trackToken.SelectToken(paramConfig.TrackTimeName)?.Value<DateTime?>());
// 轨迹发生时区 // 轨迹发生时区
var trackTimeZone = (float?) trackIsoTime?.Offset.TotalHours ?? (paramConfig.TrackTimeZoneName == null var trackTimeZone = (float?) trackIsoTime?.Offset.TotalHours ?? (paramConfig.TrackTimeZoneName == null
? null ? null
: trackToken.SelectToken(paramConfig.TrackTimeZoneName)?.Value<float>()); : trackToken.SelectToken(paramConfig.TrackTimeZoneName)?.Value<float?>());
// 轨迹发生地 // 轨迹发生地
var trackPlace = paramConfig.TrackPlaceName == null var trackPlace = paramConfig.TrackPlaceName == null
? null ? null
: trackToken.SelectToken(paramConfig.TrackPlaceName)?.Value<string>(); : trackToken.SelectToken(paramConfig.TrackPlaceName)?.Value<string?>();
// 轨迹扩展信息,根据输入的多个字段提取报文内容,构建一个新的对象的json形式 // 轨迹扩展信息,根据输入的多个字段提取报文内容,构建一个新的对象的json形式
var trackExtra = paramConfig.TrackExtraNames == null var trackExtra = paramConfig.TrackExtraNames == null
? null ? null