diff --git a/src/SJZY.Track.Api/Program.cs b/src/SJZY.Track.Api/Program.cs index ecfd4dff..8d7f64af 100644 --- a/src/SJZY.Track.Api/Program.cs +++ b/src/SJZY.Track.Api/Program.cs @@ -2,21 +2,10 @@ using SJZY.Expand.ABP.Core.Configuration; using SJZY.Expand.ABP.Core.Configuration.AutoRegistOptions; using SJZY.Expand.ABP.Core.Logger; using SJZY.Track.Api; -using SJZY.Track.Application; try { var builder = WebApplication.CreateBuilder(args); - if (args.Length > 0) - { - _BUGOptions.EnablePref = "on".Equals(args[0]); - Console.WriteLine($"是否启动优化模式：{_BUGOptions.EnablePref}"); - if (_BUGOptions.EnablePref) - { - var loadSuccess = Confluent.Kafka.Library.Load(null); - Console.WriteLine($"加载{loadSuccess},版本号：{Confluent.Kafka.Library.VersionString}"); - } - } builder .AddConfigurationContainer() .AddOptionsFromConfiguration() diff --git a/src/SJZY.Track.Api/SJZY.Track.Api.csproj b/src/SJZY.Track.Api/SJZY.Track.Api.csproj index be47c0cf..ff97789d 100644 --- a/src/SJZY.Track.Api/SJZY.Track.Api.csproj +++ b/src/SJZY.Track.Api/SJZY.Track.Api.csproj @@ -1,6 +1,6 @@  - + net7.0 @@ -10,17 +10,18 @@ - - - - - + + + + + - - + + + + + - - diff --git a/src/SJZY.Track.Api/kafkalib/librdkafka++.so.1 b/src/SJZY.Track.Api/kafkalib/librdkafka++.so.1 new file mode 100644 index 00000000..999f69c9 Binary files /dev/null and b/src/SJZY.Track.Api/kafkalib/librdkafka++.so.1 differ diff --git a/src/SJZY.Track.Api/kafkalib/librdkafka.so.1 b/src/SJZY.Track.Api/kafkalib/librdkafka.so.1 new file mode 100644 index 00000000..f9fc0367 Binary files /dev/null and b/src/SJZY.Track.Api/kafkalib/librdkafka.so.1 differ diff --git a/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs b/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs index 804ad6d5..22e5fd37 100644 --- a/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs +++ b/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs @@ -135,6 +135,11 @@ public class TrackChangeEto /// public DateTime? EstimatedDeliveryDate { get; set; } + /// + /// 目的地时区 + /// + public float? DestinationTimeZone { get; set; } + /// /// 扩展信息字段，用与轨迹推送的附加字段 /// diff --git a/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj b/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj index e8268247..23c2141f 100644 --- a/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj +++ b/src/SJZY.Track.Application.Contract/SJZY.Track.Application.Contract.csproj @@ -5,7 +5,7 @@ net7.0 True - 7.0.20 + 7.0.22 True diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index 33d65d4f..79e6998d 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs @@ -6,7 +6,6 @@ using SJZY.Track.Application.Contract.Dtos.ProvideServices; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Kafka; -using SJZY.Track.Application.Services.Warning.Handlers; using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; @@ -25,8 +24,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IBroadcastingService _broadcastingService; - // private readonly WarningSignalPublisher _warningSignalPublisher; - private readonly IServiceProvider _serviceProvider; + private readonly WarningSignalPublisher _warningSignalPublisher; public TrackDeleteEventHandler( IProvideQueries provideQueries, @@ -35,8 +33,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe ITrackDetailsEsRepository trackDetailsEsRepository, IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, - IServiceProvider serviceProvider - //WarningSignalPublisher warningSignalPublisher + WarningSignalPublisher warningSignalPublisher ) { _provideQueries = provideQueries; @@ -45,8 +42,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe _trackDetailsEsRepository = trackDetailsEsRepository; _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; - _serviceProvider = serviceProvider; - //_warningSignalPublisher = warningSignalPublisher; + _warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)] @@ -327,6 +323,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe _broadcastingService.TrackChangePublicTopic(new TrackChangeEto { EstimatedDeliveryDate = track.EstimatedDeliveryDate, + DestinationTimeZone = track.DestinationTimeZone, TrackPlacePostcode = track.TrackPlacePostcode, TrackPlaceNation = track.TrackPlaceNation, TrackPlaceNationId = track.TrackPlaceNationId, @@ -354,7 +351,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe OrderNo = track.OrderNo, OrderIdentifiers = new[] { - new OrderIdentifier() + new OrderIdentifier { OrderId = track.OrderId, OrderNo = track.OrderNo, @@ -368,43 +365,8 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe } } - private void WarningCalc_DeletedTrack(List delTracks) { - if (_BUGOptions.EnablePref) - { - var scope = _serviceProvider.CreateScope(); - var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); - _warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks); - scope.Dispose(); - return; - } - - //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹删除, delTracks); - var signals = WarningSignal.FromTrack(delTracks, WarningSignalType.轨迹删除); - - _ = Task.Run(async () => - { - var scope = _serviceProvider.CreateAsyncScope(); - try - { - var handler = scope.ServiceProvider.GetRequiredService(); - var handler2 = scope.ServiceProvider.GetRequiredService(); - var handler3 = scope.ServiceProvider.GetRequiredService(); - - await Task.Delay(1500); //确保落盘 - await handler.TrackDeletedHandleAsync(signals); - await handler2.TrackDeletedHandleAsync(signals); - await handler3.TrackUpdatedOrDeletedHandleAsync(signals); - } - catch (Exception ex) - { - LogHelper.Error(ex); - } - finally - { - await scope.DisposeAsync(); - } - }); + _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹删除, delTracks); } } diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 628c4d17..c6de6489 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs @@ -463,7 +463,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (string.IsNullOrWhiteSpace(compareFields)) return trackIdentifies.Contains($"{code}{timeStr}"); - var extra = x.GetValueOrDefault("extra"); + var extra = x.GetValueOrDefault("extra"); if (extra == null) return false; if (((JsonElement) extra).TryGetProperty(compareFields, out var comparerValueJsonElement)) @@ -538,15 +538,22 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe if (config.DateTimeFormatType == (int) DateTimeFormatEnum.Timestamp) return DateTimeHelper.ConvertToUnixTimestamp(dateTime).ToString(); - if (config.DateTimeFormatType == (int) DateTimeFormatEnum.DateTimeAndTimezone - && !string.IsNullOrEmpty(config.DateTimeFormat) - && !string.IsNullOrEmpty(config.TimezoneFormat)) + if (config.DateTimeFormatType == (int) DateTimeFormatEnum.DateTimeAndTimezone) { - var zoneStr = DateTimeHelper.TimezoneFormat(timezone, config.TimezoneFormat); - var datetimeStr = DateTimeHelper.FormatDateTime(dateTime, config.DateTimeFormat); - return $"{datetimeStr} {zoneStr}"; + // 只有时间 + if (!string.IsNullOrWhiteSpace(config.DateTimeFormat) && string.IsNullOrWhiteSpace(config.TimezoneFormat)) + return DateTimeHelper.FormatDateTime(dateTime, config.DateTimeFormat); + + // 时间和时区 + if (!string.IsNullOrWhiteSpace(config.DateTimeFormat) && !string.IsNullOrWhiteSpace(config.TimezoneFormat)) + { + var zoneStr = DateTimeHelper.TimezoneFormat(timezone, config.TimezoneFormat); + var datetimeStr = DateTimeHelper.FormatDateTime(dateTime, config.DateTimeFormat); + return $"{datetimeStr} {zoneStr}"; + } } + // 默认为 时间+时区 return dateTime.ToString("yyyy-MM-ddTHH:mm:ss.fffK"); } diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 65070319..61cd543b 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -6,7 +6,6 @@ using SJZY.OTel; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Kafka; -using SJZY.Track.Application.Services.Warning.Handlers; using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; @@ -26,11 +25,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; private readonly IBroadcastingService _broadcastingService; private readonly ILogger _logger; - private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository; - - //private readonly WarningSignalPublisher _warningSignalPublisher; - private readonly IServiceProvider _serviceProvider; + private readonly WarningSignalPublisher _warningSignalPublisher; public TrackUpsertEventHandler( IOrderRepository orderRepository, @@ -40,8 +36,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe IOrderOperationRecordEsRepository orderOperationRecordEsRepository, IBroadcastingService broadcastingService, ILogger logger, - IServiceProvider serviceProvider - //WarningSignalPublisher warningSignalPublisher + WarningSignalPublisher warningSignalPublisher ) { _orderRepository = orderRepository; @@ -51,8 +46,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe _orderOperationRecordEsRepository = orderOperationRecordEsRepository; _broadcastingService = broadcastingService; _logger = logger; - _serviceProvider = serviceProvider; - //_warningSignalPublisher = warningSignalPublisher; + _warningSignalPublisher = warningSignalPublisher; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] @@ -102,11 +96,10 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe // 不同子系统需要做内容映射 if (@event.SubSystemId != nextEvent.SubSystemId) { - var mapCode = (await _provideQueries - .MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode)) + var mapCode = (await _provideQueries.MapTrackCodesAsync(@event.SubSystemId, nextEvent.SubSystemId, @event.TrackCode)) .FirstOrDefault(); - if (mapCode == default) + if (mapCode == null) { // 跨子系统没做映射的轨迹不处理 continue; @@ -518,75 +511,12 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private void WarningCalc_AddedTrack(List tracks) { - if (_BUGOptions.EnablePref) - { - var scope = _serviceProvider.CreateScope(); - var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); - _warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹新增, tracks); - scope.Dispose(); - return; - } - - //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹新增, tracks); - var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹新增); - _ = Task.Run(async () => - { - var scope = _serviceProvider.CreateAsyncScope(); - try - { - var handler = scope.ServiceProvider.GetRequiredService(); - var handler2 = scope.ServiceProvider.GetRequiredService(); - - await Task.Delay(1500); //确保落盘 - await handler.TrackAddedHandleAsync(signals); - await handler2.TrackAddedOrUpdatedHandleAsync(signals); - } - catch (Exception ex) - { - LogHelper.Error(ex); - } - finally - { - await scope.DisposeAsync(); - } - }); + _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹新增, tracks); } private void WarningCalc_UpdatedTrack(List tracks) { - if (_BUGOptions.EnablePref) - { - var scope = _serviceProvider.CreateScope(); - var _warningSignalPublisher = scope.ServiceProvider.GetRequiredService(); - _warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks); - scope.Dispose(); - return; - } - - //_warningSignalPublisher.PublishWarningSignal(WarningSignalType.轨迹更新, tracks); - var signals = WarningSignal.FromTrack(tracks, WarningSignalType.轨迹更新); - - _ = Task.Run(async () => - { - var scope = _serviceProvider.CreateAsyncScope(); - try - { - var handler = scope.ServiceProvider.GetRequiredService(); - var handler2 = scope.ServiceProvider.GetRequiredService(); - - await Task.Delay(1500); //确保落盘 - await handler.TrackAddedOrUpdatedHandleAsync(signals); - await handler2.TrackUpdatedOrDeletedHandleAsync(signals); - } - catch (Exception ex) - { - LogHelper.Error(ex); - } - finally - { - await scope.DisposeAsync(); - } - }); + _warningSignalPublisher.PublishSignal(WarningSignalType.轨迹更新, tracks); } } diff --git a/src/SJZY.Track.Application/Hangfire/NotifyConfig.cs b/src/SJZY.Track.Application/Hangfire/NotifyConfig.cs deleted file mode 100644 index 861be32b..00000000 --- a/src/SJZY.Track.Application/Hangfire/NotifyConfig.cs +++ /dev/null @@ -1,63 +0,0 @@ -namespace SJZY.Track.Application.Hangfire; - -// 发送消息接口参数 -// https://centerapi.test.shijizhongyun.com/userApi/index.html?urls.primaryName=%E6%B6%88%E6%81%AF%E8%AE%A2%E9%98%85%E5%AE%9E%E4%BE%8B -/// MsgSubscribeInstance/SendMessageByTemplate -public class NotifyConfig -{ - private const string ContentKey = "content"; //内容参数 - - public bool IsNeedToHandle { get; set; } = true; //是否需要处理(默认为未处理,传false为无需处理,针对站内信是否处理功能) - public int MsgSource { get; set; } = 1; // 消息来源：1.消息 2.任务 3.流程 4.消息提醒 - public int SystemId { get; set; } = 9498; //子系统id 业务中台 - public string TemplateCode { get; set; } = "TrackWarningNotify"; //模板code - public List Channels { get; set; } //1:站内信 2.飞书 4.邮箱 8.短信 16.语音播报 - public List Variables { get; set; } //模板参数 - public List Receivers { get; set; } - - public sealed record Variable(string Key, string Value); - - public sealed record Receiver( - int ReceiverId, //接收人id : 内部用户Id,外部用户Id,或只有邮箱跟短信时不需要接收人Id - int RecipientType = 1, //接收人类型 (1:内部用户 2.外部用户) - string? ReceiverName = null, //接收人姓名 - string? Mail = null, //邮箱 - string? MobilePhone = null //手机号 - ); - - public static NotifyConfig CreateConfig(int userId, string message, List notifyModes) - { - return new NotifyConfig - { - Channels = notifyModes, - Receivers = new List(1) - { - new Receiver(userId) - }, - Variables = new List(1) - { - new Variable(ContentKey, message) - } - }; - } - - public static NotifyConfig CreateConfig(int userId, string message) - { - return new NotifyConfig - { - Channels = new List(2) - { - 1, - 2 - }, - Receivers = new List - { - new Receiver(userId) - }, - Variables = new List - { - new Variable(ContentKey, message) - } - }; - } -} diff --git a/src/SJZY.Track.Application/Hangfire/WarningNotifyAggregator.cs b/src/SJZY.Track.Application/Hangfire/WarningNotifyAggregator.cs index ca1d81c3..c49df4aa 100644 --- a/src/SJZY.Track.Application/Hangfire/WarningNotifyAggregator.cs +++ b/src/SJZY.Track.Application/Hangfire/WarningNotifyAggregator.cs @@ -1,6 +1,5 @@ using Microsoft.Extensions.Options; using SJZY.Track.Application.Contract.Dtos.Warning; -using SJZY.Track.Application.Helpers; using SJZY.Track.Application.Options; using SJZY.Track.Application.Services.Warning.Model; using SJZY.Track.Domain.Aggregates.KPIAreaAggregate; @@ -43,7 +42,7 @@ public sealed class WarningNotifyAggregator : ITransientDependency // 2.以人的维度聚合通知信息，并生成配置参数NotifyConfig // 3.构建并调度一个调用通知接口的立即执行任务 // 4.绑定这个任务id(可能是多个)和jobId用户跟踪任务状态 - public async Task<(List NotifyConfigs, HashSet taskIds)> ScanNotifyAndAggregateAsync() + public async Task<(List NotifyConfigs, HashSet taskIds)> ScanNotifyAndAggregateAsync() { // 构建查询游标 todo:全量查询需要优化，直接存缓存？ // var options = _optionsMonitor.CurrentValue; @@ -210,7 +209,7 @@ public sealed class WarningNotifyAggregator : ITransientDependency // 2.检查明细中为人工介入的状态 var detailsIdSet = filtered.Select(x => x.WarningDetailsId).ToHashSet(); - var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO() + var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO { DetailsIds = detailsIdSet, Status = new HashSet(1) @@ -261,10 +260,9 @@ public sealed class WarningNotifyAggregator : ITransientDependency return result; } - private List BuildNotifyConfigs(List needNotifyTasks) + private List BuildNotifyConfigs(List needNotifyTasks) { - var systemId = _optionsMonitor.CurrentValue.MiddleSystemId; - var result = new List(); + var result = new List(); if (needNotifyTasks.Count == 0) return result; // 1.以人的维度对不同类别的预警通知分组聚合 @@ -293,109 +291,51 @@ public sealed class WarningNotifyAggregator : ITransientDependency } } - // 2.使用消息模板构建通知消息，并发送通知 + // 2.使用消息模板构建通知参数，稍后排队发送通知 foreach (var (userId, typeNotifyTaskDict) in userIdNotifyTaskDict) { - var notifyMessage = BuildNotifyMessage(typeNotifyTaskDict); //聚合时出现 用户a //接收回传预警，命中规则1，通知方式(飞书,站内信) ，命中规则2，通知方式(飞书) //接收运输预警，命中规则1，通知方式(飞书,站内信) ，命中规则2，通知方式(飞书) //接收完整预警，命中规则1，通知方式(飞书) ，命中规则2，通知方式(飞书，站内信) //此时聚合后，消息为：总共6条预警消息,其中回传xx条，运输xx条，完整xx条 //同时使用飞书和站内信两种方式通知 - var config = NotifyConfig.CreateConfig(userId, notifyMessage, new List - { - 1, - 2 - }); - config.SystemId = systemId; - result.Add(config); + result.Add(BuildNotifyParam(userId, typeNotifyTaskDict)); } return result; } - private string BuildNotifyMessage(Dictionary> typeNotifyTaskDict) + private WarningNotifyParam BuildNotifyParam(int userId, Dictionary> typeNotifyTaskDict) { var options = _optionsMonitor.CurrentValue; var hours = options.ScanWarningNotifyTaskHourInterval; - var transportCount = 0; - var transportLastNHoursCount = 0; - var returnCount = 0; - var returnLastNHoursCount = 0; - var completeCount = 0; - var completeLastNHoursCount = 0; - Func hoursNBefore = x => DateTime.Now - x.CreateTime <= TimeSpan.FromHours(hours); + var param = new WarningNotifyParam + { + UserId = userId, + Hours = hours + }; + // tips:获取过去x小时的增量 + Func hoursNBefore = x => DateTime.Now - x.CreateTime <= TimeSpan.FromHours(hours); if (typeNotifyTaskDict.TryGetValue(WarningRuleType.运输时效预警.GetHashCode(), out var transportTimelinessNotifyTasks)) { - transportLastNHoursCount = transportTimelinessNotifyTasks.Count(hoursNBefore); - transportCount = transportTimelinessNotifyTasks.Count; + param.TransportLastNHoursCount = transportTimelinessNotifyTasks.Count(hoursNBefore); + param.TransportCount = transportTimelinessNotifyTasks.Count; } if (typeNotifyTaskDict.TryGetValue(WarningRuleType.回传时效预警.GetHashCode(), out var returnTimelinessNotifyTasks)) { - returnLastNHoursCount = returnTimelinessNotifyTasks.Count(hoursNBefore); - returnCount = returnTimelinessNotifyTasks.Count; + param.ReturnLastNHoursCount = returnTimelinessNotifyTasks.Count(hoursNBefore); + param.ReturnCount = returnTimelinessNotifyTasks.Count; } if (typeNotifyTaskDict.TryGetValue(WarningRuleType.完整预警.GetHashCode(), out var completeWarningNotifyTasks)) { - completeLastNHoursCount = completeWarningNotifyTasks.Count(hoursNBefore); - completeCount = completeWarningNotifyTasks.Count; + param.CompleteLastNHoursCount = completeWarningNotifyTasks.Count(hoursNBefore); + param.CompleteCount = completeWarningNotifyTasks.Count; } - return NotifyMessageBuilder.Build( - hours, - transportCount, - transportLastNHoursCount, - returnCount, - returnLastNHoursCount, - completeCount, - completeLastNHoursCount - ); - } -} - -// 标题：预警通知 -// 内容：尊敬的用户，截止到2024-12-13 10:00:01(UTC +8:00)，轨迹中心监测到与您相关的预警风险信息共计299 条，请及时处理，详情如下： -// 运输时效预警共计100条，其中过去4小时内新增的预警10条，查看详情 -// 回传时效预警共计99条，其中过去4小时内新增的预警10条，查看详情 -// 轨迹完整预警共计100条，其中过去4小时内新增的预警10条，查看详情 -public static class NotifyMessageBuilder -{ - public const string Title = "预警通知"; - - public static string Build(int hours, - int transportCount, int transportLastNHoursCount, - int returnCount, int returnLastNHoursCount, - int completeCount, int completeLastNHoursCount) - { - var dateTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"); - var totalCount = transportCount + returnCount + completeCount; - // todo:无法使用a标签填充（可能是为了安全考虑），需要具体给定一个样式 - var url1 = ""; - var url2 = ""; - var url3 = ""; - - var message = $"{Title}\n尊敬的用户，截止到{dateTime}({BuildLocalTimezoneStr()})，轨迹中心监测到与您相关的预警风险信息共计{totalCount} 条，请及时处理，详情如下："; - - if (transportCount > 0) - message += $"\n运输时效预警共计{transportCount}条，其中过去{hours}小时内新增的预警{transportLastNHoursCount}条"; - if (returnCount > 0) - message += $"\n回传时效预警共计{returnCount}条，其中过去{hours}小时内新增的预警{returnLastNHoursCount}条"; - if (completeCount > 0) - message += $"\n轨迹完整预警共计{completeCount}条，其中过去{hours}小时内新增的预警{completeLastNHoursCount}条"; - - return message; - } - - // 构建本地时区字符串标识 e.g UTC +8:00 - private static string BuildLocalTimezoneStr() - { - var offset = TimeZoneInfo.Local.BaseUtcOffset; - var flag = offset.Hours > 0 ? "+" : ""; - var offsetStr = offset.ToString().RemovePostFix(":00"); - return $"UTC {flag}{offsetStr}"; + return param; } } diff --git a/src/SJZY.Track.Application/Hangfire/WarningNotifyBuilder.cs b/src/SJZY.Track.Application/Hangfire/WarningNotifyBuilder.cs new file mode 100644 index 00000000..997d3533 --- /dev/null +++ b/src/SJZY.Track.Application/Hangfire/WarningNotifyBuilder.cs @@ -0,0 +1,122 @@ +using FeishuNetSdk; +using FeishuNetSdk.Im; +using FeishuNetSdk.Im.Dtos; +using SJZY.UserCenter.SharedModel.Dtos.Message; +using SJZY.UserCenter.SharedModel.Enums; + +namespace SJZY.Track.Application.Hangfire; + +// 标题：预警通知 +// 内容：尊敬的用户，截止到2024-12-13 10:00:01(UTC +8:00)，轨迹中心监测到与您相关的预警风险信息共计299 条，请及时处理，详情如下： +// 运输时效预警共计100条，其中过去4小时内新增的预警10条，查看详情 +// 回传时效预警共计99条，其中过去4小时内新增的预警10条，查看详情 +// 轨迹完整预警共计100条，其中过去4小时内新增的预警10条，查看详情 +public static class WarningNotifyBuilder +{ + private const string Title = "预警通知"; + + // 构建站内信发消息请求体 + public static SendMessageByTemplateInput BuildZhanNeiXinMessage(WarningNotifyParam param, int systemId) + { + const string templateCode = "TrackWarningNotify"; // 模板code + const string contentKey = "content"; //内容参数 + var dateTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"); + var totalCount = param.TransportCount + param.ReturnCount + param.CompleteCount; + + // todo:无法使用a标签填充（可能是为了安全考虑），需要具体给定一个样式 + var url1 = ""; + var url2 = ""; + var url3 = ""; + + var content = $"{Title}\n尊敬的用户，截止到{dateTime}({BuildLocalTimezoneStr()})，轨迹中心监测到与您相关的预警风险信息共计{totalCount} 条，请及时处理，详情如下："; + + if (param.TransportCount > 0) + content += $"\n运输时效预警共计{param.TransportCount}条，其中过去{param.Hours}小时内新增的预警{param.TransportLastNHoursCount}条"; + if (param.ReturnCount > 0) + content += $"\n回传时效预警共计{param.ReturnCount}条，其中过去{param.Hours}小时内新增的预警{param.ReturnLastNHoursCount}条"; + if (param.CompleteCount > 0) + content += $"\n轨迹完整预警共计{param.CompleteCount}条，其中过去{param.Hours}小时内新增的预警{param.CompleteLastNHoursCount}条"; + + // 构建用户中心发消息请求体对象 + var dto = new SendMessageByTemplateInput + { + IsNeedToHandle = true, + MsgSource = MsgSourceEnum.消息, + SystemId = systemId, + templateCode = templateCode, + channels = new List(1) + { + ChannelEnum.站内信 + }, + variables = new List(1) + { + new VariableDto + { + Key = contentKey, + Value = content + } + }, + receivers = new List(1) + { + new MsgInstanceReceiverInput + { + RecipientType = 1, //接收人类型 (1:内部用户 2.外部用户) + ReceiverId = param.UserId, //接收人id + ReceiverName = null, //接收人名称 + Mail = null, // 邮箱 + MobilePhone = null, // 手机号 + } + }, + }; + + return dto; + } + + // 构建飞书卡片发消息请求体 + public static PostImV1MessagesBodyDto BuildFeiShuCard(WarningNotifyParam param, string fsUserId) + { + var totalCount = param.TransportCount + param.ReturnCount + param.CompleteCount; + var body = new List + { + new MarkdownElement(Content: $"尊敬的用户，截止到{DateTime.Now:yyyy-MM-dd HH:mm:ss}({BuildLocalTimezoneStr()})，轨迹中心监测到与您相关的预警风险信息共计{totalCount} 条，请及时处理，详情如下：", TextAlign: "left", TextSize: "normal_v2") + }; + // todo:补充点击链接 + var url1 = ""; + var url2 = ""; + var url3 = ""; + if (param.TransportCount > 0) + body.Add(new MarkdownElement(Content: $"运输时效预警共计{param.TransportCount}条，其中过去{param.Hours}小时内新增的预警{param.TransportLastNHoursCount}条 [查看详情]({url1})", TextAlign: "left", TextSize: "normal_v2")); + if (param.ReturnCount > 0) + body.Add(new MarkdownElement(Content: $"轨迹回传时效预警共计{param.ReturnCount}条，其中过去{param.Hours}小时内新增的预警{param.ReturnLastNHoursCount}条 [查看详情]({url2})", TextAlign: "left", TextSize: "normal_v2")); + if (param.CompleteCount > 0) + body.Add(new MarkdownElement(Content: $"轨迹完整预警共计{param.CompleteCount}条，其中过去{param.Hours}小时内新增的预警{param.CompleteLastNHoursCount}条 [查看详情]({url3})", TextAlign: "left", TextSize: "normal_v2")); + + var dto = new PostImV1MessagesBodyDto + { + MsgType = "interactive", + ReceiveId = fsUserId + }; + dto.SetContent(new ElementsCardV2Dto + { + // 配置卡片全局行为 + Config = new ElementsCardV2Dto.ConfigSuffix(), + Header = new ElementsCardV2Dto.HeaderSuffix() + { + Title = new HeaderTitleElement(Title), + Template = "red" + }, + Body = new ElementsCardV2Dto.BodySuffix(body.ToArray()), + }); + + return dto; + } + + // 构建本地时区字符串标识 e.g UTC +8:00 + private static string BuildLocalTimezoneStr() + { + var offset = TimeZoneInfo.Local.BaseUtcOffset; + var flag = offset.Hours > 0 ? "+" : ""; + var offsetStr = offset.ToString().RemovePostFix(":00"); + return $"UTC {flag}{offsetStr}"; + } +} diff --git a/src/SJZY.Track.Application/Hangfire/WarningNotifyParam.cs b/src/SJZY.Track.Application/Hangfire/WarningNotifyParam.cs new file mode 100644 index 00000000..c4b1de4b --- /dev/null +++ b/src/SJZY.Track.Application/Hangfire/WarningNotifyParam.cs @@ -0,0 +1,23 @@ +namespace SJZY.Track.Application.Hangfire; + +/// +/// 构建轨迹预警通知消息所需的参数 +/// +public class WarningNotifyParam +{ + public int UserId { get; set; } + + public int Hours { get; set; } + + public int TransportCount { get; set; } + + public int TransportLastNHoursCount { get; set; } + + public int ReturnCount { get; set; } + + public int ReturnLastNHoursCount { get; set; } + + public int CompleteCount { get; set; } + + public int CompleteLastNHoursCount { get; set; } +} diff --git a/src/SJZY.Track.Application/Hangfire/WarningTaskService.cs b/src/SJZY.Track.Application/Hangfire/WarningTaskService.cs index defaf532..050d461d 100644 --- a/src/SJZY.Track.Application/Hangfire/WarningTaskService.cs +++ b/src/SJZY.Track.Application/Hangfire/WarningTaskService.cs @@ -1,3 +1,4 @@ +using FeishuNetSdk; using Hangfire; using Microsoft.Extensions.Options; using SJZY.Expand.ABP.Core.Logger; @@ -11,9 +12,8 @@ using SJZY.Track.Domain.Aggregates.WarningAggregate.Return; using SJZY.Track.Domain.Dtos.Warning; using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Extensions; +using SJZY.Track.Domain.Queries; using SJZY.UserCenter.HttpApi.Services.Message; -using SJZY.UserCenter.SharedModel.Dtos.Message; -using SJZY.UserCenter.SharedModel.Enums; using Volo.Abp.DependencyInjection; using Volo.Abp.Http.Client.DynamicProxying; @@ -21,8 +21,8 @@ namespace SJZY.Track.Application.Hangfire; public interface IWarningTaskService { - [AutomaticRetry(Attempts = 3)] - Task CallNotifyAsync(NotifyConfig config, CancellationToken ct); + [AutomaticRetry(Attempts = 1)] + Task CallNotifyAsync(WarningNotifyParam param, CancellationToken ct); [AutomaticRetry(Attempts = 3, ExceptOn = new[] { typeof(BusinessException) })] Task CallTrackRetryPushAsync(ReturnWarningRetryPushTaskEntity entity, CancellationToken ct); @@ -30,51 +30,57 @@ public interface IWarningTaskService public class WarningTaskService : IWarningTaskService, ITransientDependency { + private readonly IOptionsMonitor _interfaceRelayOptions; + private readonly IOptionsMonitor _warningTaskOptions; private readonly IHttpClientProxy _requestSendMessageService; private readonly IHttpClientProxy _interfaceForwardServiceV2; - private readonly IOptionsMonitor _options; + private readonly IFeishuTenantApi _feishuTenantApi; private readonly IWarningDetailsEsRepository _warningDetailsEsRepository; private readonly IWarningTaskRepository _warningTaskRepository; + private readonly IVFsUserQueries _vfsUserQueries; - public WarningTaskService(IHttpClientProxy interfaceForwardServiceV2, IOptionsMonitor options, IHttpClientProxy requestSendMessageService, IWarningDetailsEsRepository warningDetailsEsRepository, IWarningTaskRepository warningTaskRepository) + public WarningTaskService( + IOptionsMonitor interfaceRelayOptions, + IOptionsMonitor warningTaskOptions, + IHttpClientProxy interfaceForwardServiceV2, + IHttpClientProxy requestSendMessageService, + IFeishuTenantApi feishuTenantApi, + IWarningDetailsEsRepository warningDetailsEsRepository, + IWarningTaskRepository warningTaskRepository, + IVFsUserQueries vfsUserQueries) { + _interfaceRelayOptions = interfaceRelayOptions; + _warningTaskOptions = warningTaskOptions; _interfaceForwardServiceV2 = interfaceForwardServiceV2; _requestSendMessageService = requestSendMessageService; + _feishuTenantApi = feishuTenantApi; _warningDetailsEsRepository = warningDetailsEsRepository; _warningTaskRepository = warningTaskRepository; - _options = options; + _vfsUserQueries = vfsUserQueries; } #region 用户中心调用，通知（一次） - [AutomaticRetry(Attempts = 3)] - public async Task CallNotifyAsync(NotifyConfig config, CancellationToken ct) + [AutomaticRetry(Attempts = 1)] + public async Task CallNotifyAsync(WarningNotifyParam param, CancellationToken ct) { ct.ThrowIfCancellationRequested(); - var r = await _requestSendMessageService.Service.SendMessageByTemplateAsync(new SendMessageByTemplateInput - { - IsNeedToHandle = config.IsNeedToHandle, - MsgSource = (MsgSourceEnum) config.MsgSource, - SystemId = config.SystemId, - templateCode = config.TemplateCode, - channels = config.Channels.Select(x => (ChannelEnum) x).ToList(), - variables = config.Variables.Select(x => new VariableDto() - { - Key = x.Key, - Value = x.Value - }).ToList(), - receivers = config.Receivers.Select(x => new MsgInstanceReceiverInput() - { - ReceiverId = x.ReceiverId, - RecipientType = x.RecipientType - }).ToList(), - }); + var options = _warningTaskOptions.CurrentValue; + // 发送站内信 + // https://centerapi.test.shijizhongyun.com/userApi/index.html?urls.primaryName=%E6%B6%88%E6%81%AF%E8%AE%A2%E9%98%85%E5%AE%9E%E4%BE%8B + // MsgSubscribeInstance/SendMessageByTemplate + var r = await _requestSendMessageService.Service.SendMessageByTemplateAsync(WarningNotifyBuilder.BuildZhanNeiXinMessage(param, options.MiddleSystemId)); if (r.Code != GlobalConst.SuccessCode) - { - throw new BusinessException(message: $"用户中心调用失败：SendMessageByTemplateAsync" + - $"Msg:{r.Msg}"); - } + throw new BusinessException(message: $"用户中心调用失败：SendMessageByTemplateAsync Msg:{r.Msg}"); + + // 发送FeiShu卡片信息 + var fsUserId = await _vfsUserQueries.GetFirstAsync(x => x.UserCenterId == param.UserId, y => y.FsEmployeeId); + if (string.IsNullOrWhiteSpace(fsUserId)) + throw new BusinessException(message: "查询用户中心失败：SendMessageByTemplateAsync Msg:未找到用户对应的飞书用户Id"); + var res = await _feishuTenantApi.PostImV1MessagesAsync("user_id", WarningNotifyBuilder.BuildFeiShuCard(param, fsUserId)); + if (res.Code != 0) + throw new BusinessException(message: $"飞书调用失败：SendMessageByTemplateAsync Msg:{res.Msg}"); } #endregion @@ -107,8 +113,8 @@ public class WarningTaskService : IWarningTaskService, ITransientDependency using (Track2InterfaceRelaySubSystemContext.Use(new Track2InterfaceRelaySubSystemContext(task.SubSystemCode))) { var upStreamCode = isAdd - ? _options.CurrentValue.PushTrackInterfaceCode - : _options.CurrentValue.TrackUpdateInterfaceCode; + ? _interfaceRelayOptions.CurrentValue.PushTrackInterfaceCode + : _interfaceRelayOptions.CurrentValue.TrackUpdateInterfaceCode; var r = await _interfaceForwardServiceV2.Service.BatchForwardThenPushToQueueAsync( upStreamCode: upStreamCode, diff --git a/src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs b/src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs index c1f0e04e..9aaaff06 100644 --- a/src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs +++ b/src/SJZY.Track.Application/Kafka/WarningCalcBackgroundService.cs @@ -1,7 +1,5 @@ using System.Diagnostics; using Confluent.Kafka; -using Confluent.Kafka.Core.Consumer; -using Confluent.Kafka.Core.Producer; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Options; using SJZY.Expand.ABP.Core.Logger; @@ -20,23 +18,20 @@ namespace SJZY.Track.Application.Kafka; public sealed class WarningCalcBackgroundService : BackgroundService { private readonly IServiceScopeFactory _serviceScopeFactory; - private readonly IKafkaConsumer _consumer; - private readonly IKafkaProducer _producer; - private readonly IKafkaProducer _dlxProducer; + private readonly IConsumer _consumer; + private readonly WarningSignalPublisher _warningSignalPublisher; private TrackWarningTaskOptions _options; public WarningCalcBackgroundService( IOptionsMonitor optionsMonitor, IServiceScopeFactory serviceScopeFactory, - IKafkaConsumer consumer, - IKafkaProducer producer, - IKafkaProducer dlxProducer + IConsumer consumer, + WarningSignalPublisher warningSignalPublisher ) { _serviceScopeFactory = serviceScopeFactory; _consumer = consumer; - _producer = producer; - _dlxProducer = dlxProducer; + _warningSignalPublisher = warningSignalPublisher; _options = optionsMonitor.CurrentValue; optionsMonitor.OnChange(x => _options = x); } @@ -175,7 +170,7 @@ public sealed class WarningCalcBackgroundService : BackgroundService var tasks = needRetryMetadata.Select(async metadata => { metadata.SetTrackSignalMessageHeader(); - await _producer.ProduceAsync(metadata.OwenMessage, CancellationToken.None); + await _warningSignalPublisher.PublishSignalAsync(metadata.OwenMessage, CancellationToken.None); }).ToArray(); await Task.WhenAll(tasks); @@ -193,7 +188,7 @@ public sealed class WarningCalcBackgroundService : BackgroundService { metadata.SetTrackSignalMessageHeader(); var msg = metadata.OwenMessage; - await _dlxProducer.ProduceAsync(new Message + await _warningSignalPublisher.PublishDlxSignalAsync(new Message { Key = msg.Value.OrderId, Value = msg.Value, @@ -282,6 +277,5 @@ public sealed class WarningCalcBackgroundService : BackgroundService base.Dispose(); _consumer.Close(); _consumer.Dispose(); - _producer.Dispose(); } } diff --git a/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs b/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs index 485e6ab8..8d9f2cf7 100644 --- a/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs +++ b/src/SJZY.Track.Application/Kafka/WarningCalcServiceExtension.cs @@ -1,20 +1,13 @@ -//#define ProtoBuf -#define NewtonsoftJson -using System.Globalization; using System.Net; +using System.Text; using System.Text.Json; -using System.Text.Json.Serialization; using Confluent.Kafka; -using Confluent.Kafka.Core.Consumer; -using Confluent.Kafka.Core.Producer; using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.Logging; -using Newtonsoft.Json; -using Newtonsoft.Json.Converters; -using Newtonsoft.Json.Serialization; +using Microsoft.Extensions.DependencyInjection.Extensions; using SJZY.Expand.ABP.Core.Logger; using SJZY.Track.Application.Helpers; using SJZY.Track.Application.Services.Warning.Signals; +using JsonSerializer = System.Text.Json.JsonSerializer; namespace SJZY.Track.Application.Kafka; @@ -22,6 +15,7 @@ public static class WarningCalcServiceExtension { public static void AddWarningCalcServices(this IServiceCollection services, IConfiguration conf) { + PrintKafkaLibVersion(); var env = "-" + conf.GetValue("Environment")!.ToLower(); // 预警信号批处理后台任务 services.AddHostedService(); @@ -37,140 +31,194 @@ public static class WarningCalcServiceExtension private static void RegisterConsumer(IServiceCollection services, IConfiguration conf, string groupId, string[] subTopic) { - services.AddSingleton>(sp => + services.AddJsonSerializer(); + services.AddJsonSerializer(); + services.AddSingleton>(sp => { - var bb = new KafkaConsumerBuilder() - .WithConsumerConfiguration(o => - { - o.WithClientId(Dns.GetHostName()) - .WithGroupId(groupId) - .WithBootstrapServers(conf.GetValue("Kafka:BootstrapServers")) - .WithAutoOffsetReset(AutoOffsetReset.Earliest) - //偏移提交控制 - .WithEnablePartitionEof(false) - .WithEnableAutoOffsetStore(false) - .WithEnableAutoCommit(false) - //链接控制 - .WithSessionTimeoutMs(conf.GetValue("Kafka:Consumer:SessionTimeoutMs")) - .WithMaxPollIntervalMs(conf.GetValue("Kafka:Consumer:MaxPollIntervalMs")) - .WithHeartbeatIntervalMs(conf.GetValue("Kafka:Consumer:HeartbeatIntervalMs")) - //吞吐量 - .WithFetchMinBytes(conf.GetValue("Kafka:Consumer:FetchMinBytes")) - .WithFetchMaxBytes(conf.GetValue("Kafka:Consumer:FetchMaxBytes")) - .WithFetchWaitMaxMs(conf.GetValue("Kafka:Consumer:FetchWaitMaxMs")) - //日志 - .WithDebug("consumer,cgrp,protocol") - .WithTopicSubscriptions(subTopic); - }); - - bb.WithLoggerFactory(sp.GetRequiredService()); DebounceHelper.DebounceWrapper? reportFuncWrapper = null; - bb.WithErrorHandler((consumer, error) => - { - if (error.IsFatal) + var bb = new ConsumerBuilder( + new ConsumerConfig + { + ClientId = Dns.GetHostName(), + GroupId = groupId, + BootstrapServers = conf.GetValue("Kafka:BootstrapServers"), + AutoOffsetReset = AutoOffsetReset.Earliest, + //偏移提交控制 + EnablePartitionEof = false, + EnableAutoOffsetStore = false, + EnableAutoCommit = false, + //链接控制 + SessionTimeoutMs = conf.GetValue("Kafka:Consumer:SessionTimeoutMs"), + MaxPollIntervalMs = conf.GetValue("Kafka:Consumer:MaxPollIntervalMs"), + HeartbeatIntervalMs = conf.GetValue("Kafka:Consumer:HeartbeatIntervalMs"), + //吞吐量 + FetchMinBytes = conf.GetValue("Kafka:Consumer:FetchMinBytes"), + FetchMaxBytes = conf.GetValue("Kafka:Consumer:FetchMaxBytes"), + FetchWaitMaxMs = conf.GetValue("Kafka:Consumer:FetchWaitMaxMs"), + //日志 + Debug = "consumer,cgrp,protocol", + CancellationDelayMaxMs = 100 + }) + .SetKeyDeserializer(sp.GetRequiredService>()) + .SetValueDeserializer(sp.GetRequiredService>()) + .SetErrorHandler((_, error) => { - LogHelper.Error($"Kafka致命错误${error}"); - return; - } + if (error.IsFatal) + { + LogHelper.Error($"Kafka致命错误${error}"); + return; + } - // tips: 防抖以防止高频上报错误 - reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () => - { - if (error.Code == ErrorCode.Local_Transport) - LogHelper.Error($"Kafka链接错误{error}"); - else if (error.Code == ErrorCode.Local_AllBrokersDown) - LogHelper.Error($"Kafka所有Broker都挂了{error}"); + // tips: 防抖以防止高频上报错误 + reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () => + { + if (error.Code == ErrorCode.Local_Transport) + LogHelper.Error($"Kafka链接错误{error}"); + else if (error.Code == ErrorCode.Local_AllBrokersDown) + LogHelper.Error($"Kafka所有Broker都挂了{error}"); + }); + reportFuncWrapper.Debounce(); }); - reportFuncWrapper.Debounce(); - }); -#if ProtoBuf - bb.WithProtobufNetDeserializer(); -#elif NewtonsoftJson - bb.WithNewtonsoftJsonDeserializer(builder => - builder.WithNullValueHandling(NullValueHandling.Ignore) // Ignores null values - .WithReferenceLoopHandling(ReferenceLoopHandling.Ignore) // Ignores reference loops - .WithContractResolver(new CamelCasePropertyNamesContractResolver()) // Camel case property names - ); -#else - bb.WithJsonCoreDeserializer(bc => - bc.WithReferenceHandler(ReferenceHandler.IgnoreCycles) // Handles circular references - .WithPropertyNamingPolicy(JsonNamingPolicy.CamelCase) // Uses camelCase naming - .WithDefaultIgnoreCondition(JsonIgnoreCondition.WhenWritingNull) - ); -#endif - return (bb.Build() as IKafkaConsumer)!; + var c = bb.Build(); + c.Subscribe(subTopic); + return c; }); } private static void RegisterProducer(IServiceCollection services, IConfiguration conf, string? defaultTopic = null) { - services.AddSingleton>(sp => + services.AddJsonSerializer(); + services.AddJsonSerializer(); + services.AddSingleton>(sp => { - var bb = new KafkaProducerBuilder() - .WithProducerConfiguration(o => - { - o.WithClientId(Dns.GetHostName()) - .WithBootstrapServers(conf.GetValue("Kafka:BootstrapServers")) - // 确保消息发送可靠 - .WithAcks(Acks.All) - .WithEnableIdempotence(true) // 幂等发送 - .WithMessageSendMaxRetries(conf.GetValue("Kafka:Producer:MessageSendMaxRetries")) - .WithRequestTimeoutMs(conf.GetValue("Kafka:Producer:RequestTimeoutMs")) - // 消息体大小 - .WithCompressionType(conf.GetValue("Kafka:Producer:CompressionType")) - .WithCompressionLevel(conf.GetValue("Kafka:Producer:CompressionLevel")) //Gzip 0-9 LZ4 0-12 数值越大压缩效果越好cpu占用越多 - // 消息发送速率控制(任意一条件先达到就发送) - .WithLingerMs(conf.GetValue("Kafka:Producer:LingerMs")) - .WithBatchSize(conf.GetValue("Kafka:Producer:BatchSize")) - .WithBatchNumMessages(conf.GetValue("Kafka:Producer:BatchNumMessages")) - // 发送缓冲内存限制 - .WithQueueBufferingMaxMessages(conf.GetValue("Kafka:Producer:QueueBufferingMaxMessages")) - .WithQueueBufferingMaxKbytes(conf.GetValue("Kafka:Producer:QueueBufferingMaxKbytes")) - // 分区器 一致性hash/随机 - .WithPartitioner(conf.GetValue("Kafka:Producer:Partitioner")) - .WithDefaultTopic(defaultTopic); - }); - - bb.WithLoggerFactory(sp.GetRequiredService()); DebounceHelper.DebounceWrapper? reportFuncWrapper = null; - bb.WithErrorHandler((producer, error) => - { - if (error.IsFatal) + var bb = new ProducerBuilder(new ProducerConfig() { - LogHelper.Error($"Kafka致命错误${error}"); - return; - } + ClientId = Dns.GetHostName(), + BootstrapServers = conf.GetValue("Kafka:BootstrapServers"), + // 确保消息发送可靠 + Acks = Acks.All, + EnableIdempotence = true, + MessageSendMaxRetries = conf.GetValue("Kafka:Producer:MessageSendMaxRetries"), + RequestTimeoutMs = conf.GetValue("Kafka:Producer:RequestTimeoutMs"), + // 消息体大小 + CompressionType = conf.GetValue("Kafka:Producer:CompressionType"), + CompressionLevel = conf.GetValue("Kafka:Producer:CompressionLevel"), + // 消息发送速率控制(任意一条件先达到就发送) + LingerMs = conf.GetValue("Kafka:Producer:LingerMs"), + BatchSize = conf.GetValue("Kafka:Producer:BatchSize"), + BatchNumMessages = conf.GetValue("Kafka:Producer:BatchNumMessages"), + // 发送缓冲内存限制 + QueueBufferingMaxMessages = conf.GetValue("Kafka:Producer:QueueBufferingMaxMessages"), + QueueBufferingMaxKbytes = conf.GetValue("Kafka:Producer:QueueBufferingMaxKbytes"), + // 分区器 一致性hash/随机 + Partitioner = conf.GetValue("Kafka:Producer:Partitioner"), + }) + .SetKeySerializer(sp.GetRequiredService>()) + .SetValueSerializer(sp.GetRequiredService>()) + .SetErrorHandler((_, error) => + { + if (error.IsFatal) + { + LogHelper.Error($"Kafka致命错误${error}"); + return; + } - // tips: 防抖以防止高频上报错误 - reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () => - { - if (error.Code == ErrorCode.Local_Transport) - LogHelper.Error($"Kafka链接错误{error}"); - else if (error.Code == ErrorCode.Local_AllBrokersDown) - LogHelper.Error($"Kafka所有Broker都挂了{error}"); + // tips: 防抖以防止高频上报错误 + reportFuncWrapper ??= DebounceHelper.WrapMethod(TimeSpan.FromSeconds(5), () => + { + if (error.Code == ErrorCode.Local_Transport) + LogHelper.Error($"Kafka链接错误{error}"); + else if (error.Code == ErrorCode.Local_AllBrokersDown) + LogHelper.Error($"Kafka所有Broker都挂了{error}"); + }); + reportFuncWrapper.Debounce(); }); - reportFuncWrapper.Debounce(); - }); -#if ProtoBuf - // 2000w条消息数据包压缩从 2g -> 1g - bb.WithProtobufNetSerializer(); -#elif NewtonsoftJson - bb.WithNewtonsoftJsonSerializer(builder => - builder.WithNullValueHandling(NullValueHandling.Ignore) // Ignores null values - .WithReferenceLoopHandling(ReferenceLoopHandling.Ignore) // Ignores reference loops - .WithContractResolver(new CamelCasePropertyNamesContractResolver()) // Camel case property names - ); -#else - bb.WithJsonCoreSerializer(bc => - bc.WithReferenceHandler(ReferenceHandler.IgnoreCycles) // Handles circular references - .WithPropertyNamingPolicy(JsonNamingPolicy.CamelCase) // Uses camelCase naming - .WithDefaultIgnoreCondition(JsonIgnoreCondition.WhenWritingNull) - ); -#endif - return (bb.Build() as IKafkaProducer)!; + return bb.Build(); }); } + + private static void AddJsonSerializer(this IServiceCollection services, Action? setup = null) + { + services.TryAddSingleton(_ => + { + var o = new JsonSerializerOptions(); + setup?.Invoke(o); + return new JsonCoreSerializer(o); + }); + } + + public static List> ConsumeBatch(this IConsumer consumer, int batchSize, TimeSpan timeout) + { + using var cts = new CancellationTokenSource(timeout); + var list = new List>(); + while (list.Count < batchSize && !cts.IsCancellationRequested) + { + var consumeResult = consumer.Consume(100); + if (consumeResult != null) + list.Add(consumeResult); + } + + return list; + } + + public static void PrintKafkaLibVersion() + { + var loadSuccess = Library.Load(null); + Console.WriteLine($@" + +-------------------------------------------------------+ + | | + | LibrdKafka | + | 加载：{loadSuccess}，版本号：{Library.VersionString} | + | | + +-------------------------------------------------------+ + "); + + } +} + +internal sealed class JsonCoreSerializer : ISerializer, IDeserializer +{ + private readonly Encoding _encoding; + private readonly JsonSerializerOptions _options; + + public JsonCoreSerializer(JsonSerializerOptions options) + { + _options = options ?? throw new ArgumentNullException(nameof(options)); + _encoding = Encoding.UTF8; + } + + public byte[] Serialize(T data, SerializationContext context) + { + if (data is null) + { + return null; + } + + var json = JsonSerializer.Serialize(data, data.GetType(), _options); + + var result = _encoding.GetBytes(json); + + return result; + } + + public T Deserialize(ReadOnlySpan data, bool isNull, SerializationContext context) + { + if (isNull) + { + return default; + } + +#if NETSTANDARD2_0 + var json = _encoding.GetString(data.ToArray()); +#else + var json = _encoding.GetString(data); +#endif + var result = JsonSerializer.Deserialize(json, _options); + + return result; + } } diff --git a/src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs b/src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs index 95e0336f..313705af 100644 --- a/src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs +++ b/src/SJZY.Track.Application/Kafka/WarningSignalPublisher.cs @@ -1,5 +1,5 @@ using Confluent.Kafka; -using Confluent.Kafka.Core.Producer; +using Microsoft.Extensions.Configuration; using SJZY.Track.Application.Services.Warning.Signals; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using Volo.Abp.DependencyInjection; @@ -9,27 +9,48 @@ namespace SJZY.Track.Application.Kafka; /// /// 预警信号发布者 /// -public sealed class WarningSignalPublisher : ITransientDependency +public sealed class WarningSignalPublisher : IDisposable, ISingletonDependency { - private readonly IKafkaProducer _producer; + private readonly IProducer _producer; + private readonly IProducer _dlxProducer; + private readonly string env; - public WarningSignalPublisher(IKafkaProducer producer) + public WarningSignalPublisher(IProducer producer, IProducer dlxProducer, IConfiguration conf) { _producer = producer; + _dlxProducer = dlxProducer; + env = "-" + conf.GetValue("Environment")!.ToLower(); } - public void PublishWarningSignal(WarningSignalType type, List tracks) + public void PublishSignal(WarningSignalType type, List tracks) { foreach (var track in tracks) - PublishWarningSignal(type, track); + PublishSignal(type, track); } - public void PublishWarningSignal(WarningSignalType type, TrackDetailsIndex track) + public void PublishSignal(WarningSignalType type, TrackDetailsIndex track) { - _producer.Produce(new Message + _producer.Produce(GlobalConst.WarningKafkaTopic + env, new Message { Key = new WarningSignalKey(type, track.OrderId), Value = WarningSignal.FromTrack(track, type) }); } + + public Task PublishSignalAsync(Message msg, CancellationToken ct = default) + { + return _producer.ProduceAsync(GlobalConst.WarningKafkaTopic + env, msg, ct); + } + + + public Task PublishDlxSignalAsync(Message message, CancellationToken ct = default) + { + return _dlxProducer.ProduceAsync(GlobalConst.WarningDLXKafkaTopic + env, message, ct); + } + + public void Dispose() + { + _producer.Dispose(); + _dlxProducer.Dispose(); + } } diff --git a/src/SJZY.Track.Application/SJZY.Track.Application.csproj b/src/SJZY.Track.Application/SJZY.Track.Application.csproj index fcb4b392..06efb7c6 100644 --- a/src/SJZY.Track.Application/SJZY.Track.Application.csproj +++ b/src/SJZY.Track.Application/SJZY.Track.Application.csproj @@ -9,10 +9,8 @@ - - - - + + diff --git a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs index 049f9ba0..067b34c7 100644 --- a/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs +++ b/src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs @@ -32,6 +32,7 @@ public class BroadcastingService : IBroadcastingService var data = new TrackChangeEto { EstimatedDeliveryDate = eto.EstimatedDeliveryDate, + DestinationTimeZone = eto.DestinationTimeZone, TrackPlacePostcode = eto.TrackPlacePostcode, TrackPlaceNation = eto.TrackPlaceNation, TrackPlaceNationId = eto.TrackPlaceNationId, diff --git a/src/SJZY.Track.Application/Services/TrackPush/TrackPushService.cs b/src/SJZY.Track.Application/Services/TrackPush/TrackPushService.cs index 646da5e2..8cd08d74 100644 --- a/src/SJZY.Track.Application/Services/TrackPush/TrackPushService.cs +++ b/src/SJZY.Track.Application/Services/TrackPush/TrackPushService.cs @@ -5,12 +5,14 @@ using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Contract.Services; +using SJZY.Track.Application.Helpers; using SJZY.Track.Application.Localization; using SJZY.Track.Application.Services.Warning.Handlers; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Dtos.Common; using SJZY.Track.Domain.Dtos.TrackPush; +using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; namespace SJZY.Track.Application.Services.TrackPush; @@ -112,7 +114,7 @@ public class TrackPushService : ApplicationService, ISingletonDependency /// /// [HttpPost("Edit")] - public async Task EditAsync([FromBody] EditTrackPushConfigInput input) + public async Task EditAsync([FromBody] EditTrackPushConfigInput input) { if (input.Options != OptionTypeNoNull.删除) { @@ -124,6 +126,14 @@ public class TrackPushService : ApplicationService, ISingletonDependency throw new BusinessException(message: _stringLocalizer["IsTrackPushConfigAny"]); } + var msg = string.Empty; + if (input.Options is OptionTypeNoNull.新增 or OptionTypeNoNull.修改) + { + var (success, message) = CheckDateTimeFormat(input.DateTimeFormat, input.DateTimeFormatType, input.TimezoneFormat); + if (!success) throw new BusinessException(message: message); + msg = message; + } + using var ts = TransactionScopeHelper.Create(); switch (input.Options) @@ -161,7 +171,7 @@ public class TrackPushService : ApplicationService, ISingletonDependency } ts.Complete(); - return true; + return msg; } @@ -346,4 +356,52 @@ public class TrackPushService : ApplicationService, ISingletonDependency } }); } + + private static (bool Success, string Message) CheckDateTimeFormat(string? dateTimeFormat, int? dateTimeFormatType, string? timezoneFormat) + { + const int defaultTimeZone = 8; + var dateTime = DateTime.Now; + string dateStr; + + try + { + if (dateTimeFormatType == (int) DateTimeFormatEnum.Timestamp) + { + dateStr = DateTimeHelper.ConvertToUnixTimestamp(dateTime).ToString(); + return (true, $"格式化样例：{dateStr}"); + } + + if (dateTimeFormatType == (int) DateTimeFormatEnum.DateTimeAndTimezone) + { + // 没时间有时区，出错 + if (string.IsNullOrWhiteSpace(dateTimeFormat) && !string.IsNullOrWhiteSpace(timezoneFormat)) + return (false, "如果填写了时区格式，那么时间格式是必须的"); + + // 只有时间 + if (!string.IsNullOrWhiteSpace(dateTimeFormat) && string.IsNullOrWhiteSpace(timezoneFormat)) + { + var datetimeStr = DateTimeHelper.FormatDateTime(dateTime, dateTimeFormat); + return (true, $"格式化样例：{datetimeStr}"); + } + + // 时间和时区同时存在 + if (!string.IsNullOrWhiteSpace(dateTimeFormat) && !string.IsNullOrWhiteSpace(timezoneFormat)) + { + var zoneStr = DateTimeHelper.TimezoneFormat(defaultTimeZone, timezoneFormat); + var datetimeStr = DateTimeHelper.FormatDateTime(dateTime, dateTimeFormat); + dateStr = $"{datetimeStr} {zoneStr}"; + return (true, $"格式化样例：{dateStr}"); + } + } + } + catch (Exception e) + { + return (false, $"时间格式错误请检查：{e.Message}"); + } + + // 默认情况 + dateTimeFormat ??= "yyyy-MM-ddTHH:mm:ss.fffK"; + dateStr = dateTime.ToString(dateTimeFormat); + return (true, $"格式化样例：{dateStr}"); + } } diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index a3020ec5..3a059835 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -139,7 +139,7 @@ public class SjzyTrackApplicationModule : AbpModule Console.WriteLine("-- 当前为VS本地环境，用内存存储和内存队列"); } - Action action = options => + Action action = options => { options.ConnectionString = $"amqp://{configuration["RabbitMQ:UserName"]}:{configuration["RabbitMQ:Password"]}@{configuration["RabbitMQ:HostName"]}:{configuration["RabbitMQ:Port"]}"; @@ -203,12 +203,19 @@ public class SjzyTrackApplicationModule : AbpModule $",abortConnect=false"; // 注册预警计算服务 - if (_BUGOptions.EnablePref) - context.Services.AddWarningCalcServices(configuration); + context.Services.AddWarningCalcServices(configuration); // 注册预警任务后台服务 context.Services.AddHostedService(); + // 注册飞书服务（用于预警通知） + context.Services.AddFeishuNetSdk(o => + { + o.AppId = configuration.GetValue("FeiShu:AppId")!; + o.AppSecret = configuration.GetValue("FeiShu:AppSecret")!; + o.EnableLogging = configuration.GetValue("FeiShu:EnableLogging"); + }); + // 注册HangFire服务 context.Services.Configure(configuration.GetSection("WarningTask")); context.Services.AddHangfire(o => diff --git a/src/SJZY.Track.Application/_BUGOptions.cs b/src/SJZY.Track.Application/_BUGOptions.cs deleted file mode 100644 index 3418cbe9..00000000 --- a/src/SJZY.Track.Application/_BUGOptions.cs +++ /dev/null @@ -1,6 +0,0 @@ -namespace SJZY.Track.Application; - -public static class _BUGOptions -{ - public static bool EnablePref = false; -} diff --git a/src/SJZY.Track.Domain/Queries/IVFsUserQueries.cs b/src/SJZY.Track.Domain/Queries/IVFsUserQueries.cs new file mode 100644 index 00000000..debda110 --- /dev/null +++ b/src/SJZY.Track.Domain/Queries/IVFsUserQueries.cs @@ -0,0 +1,7 @@ +using SJZY.Track.Domain.VEntities; + +namespace SJZY.Track.Domain.Queries; + +public interface IVFsUserQueries : IBasicQuery +{ +} diff --git a/src/SJZY.Track.Domain/VEntities/VFSUserEntity.cs b/src/SJZY.Track.Domain/VEntities/VFSUserEntity.cs new file mode 100644 index 00000000..71297838 --- /dev/null +++ b/src/SJZY.Track.Domain/VEntities/VFSUserEntity.cs @@ -0,0 +1,65 @@ +namespace SJZY.Track.Domain.VEntities; + +/// +/// 飞书用户表，用户中心用户关联飞书用户 +/// +[SugarTable("v_fs_user")] +public class VFSUserEntity : IEntity +{ + [SugarColumn(ColumnName = "Id", IsPrimaryKey = true, IsIdentity = true)] + public int Id { get; set; } + + /// + /// 用户中心id + /// + [SugarColumn(ColumnName = "user_center_id")] + public int UserCenterId { get; set; } + + /// + /// 用户名 + /// + [SugarColumn(ColumnName = "user_name")] + public string UserName { get; set; } + + /// + /// 用户手机号 + /// + [SugarColumn(ColumnName = "mobile")] + public string Mobile { get; set; } + + /// + /// 飞书 union id + /// + [SugarColumn(ColumnName = "union_id")] + public string UnionId { get; set; } + + /// + /// 飞书 user id + /// + [SugarColumn(ColumnName = "fs_employee_id")] + public string FsEmployeeId { get; set; } + + /// + /// 飞书头像 + /// + [SugarColumn(ColumnName = "avatar")] + public string Avatar { get; set; } + + /// + /// 自建的飞书登录应用下的OpenId + /// + [SugarColumn(ColumnName = "fsloginApp_openid")] + public string FsLoginAppOpenId { get; set; } + + /// + /// 创建时间 + /// + [SugarColumn(ColumnName = "create_time")] + public DateTime CreateTime { get; set; } + + /// + /// 更新时间 + /// + [SugarColumn(ColumnName = "update_time")] + public DateTime UpdateTime { get; set; } +} diff --git a/src/SJZY.Track.Repository/Queries/VFsUserQueries.cs b/src/SJZY.Track.Repository/Queries/VFsUserQueries.cs new file mode 100644 index 00000000..fb047384 --- /dev/null +++ b/src/SJZY.Track.Repository/Queries/VFsUserQueries.cs @@ -0,0 +1,9 @@ +using SJZY.Track.Domain.Queries; +using SJZY.Track.Domain.VEntities; + +namespace SJZY.Track.Repository.Queries; + +public class VFsUserQueries: BasicQuery, IVFsUserQueries, ITransientDependency +{ + +}