From 23a1082d80eed5b06cb8e27fa89073d9b662bfc7 Mon Sep 17 00:00:00 2001 From: eli Date: Fri, 15 Nov 2024 16:20:09 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E8=BD=A8=E8=BF=B9=E5=8F=98?= =?UTF-8?q?=E6=9B=B4=E5=B9=BF=E6=92=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Events/TrackChangeEto.cs | 111 ++++++++++++++++++ .../EventHandlers/TrackUpsertEventHandler.cs | 71 ++++++++++- .../SjzyTrackApplicationModule.cs | 97 +++++++++------ .../Consts/GlobalConst.MQ.cs | 7 ++ .../Dtos/TrackContents/TrackContentVO.cs | 5 + .../Queries/ProvideQueries.cs | 3 +- 6 files changed, 250 insertions(+), 44 deletions(-) create mode 100644 src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs diff --git a/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs b/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs new file mode 100644 index 00000000..9d3a0a49 --- /dev/null +++ b/src/SJZY.Track.Application.Contract/Events/TrackChangeEto.cs @@ -0,0 +1,111 @@ +using SJZY.Track.Application.Contract.Dtos.ProvideServices; + +namespace SJZY.Track.Application.Contract.Events; + +public class TrackChangeEto +{ + /// + /// 操作类型：1新增，2删除，3修改 + /// + public int Option { get; set; } + + /// + /// 子系统id + /// + public int SubSystemId { get; set; } + + /// + /// 客户id + /// + public int CustomId { get; set; } + + /// + /// 订单id + /// + public OrderIdentifier[] OrderIdentifiers { get; set; } = null!; + + /// + /// 操作人 + /// + public string? OperationMan { get; set; } + + /// + /// 轨迹内容代码 + /// + public string TrackCode { get; set; } = null!; + + /// + /// 轨迹内容 + /// + public string? Content { get; set; } + + /// + /// 轨迹内容英文 + /// + public string? EnStatus { get; set; } + + /// + /// 轨迹状态代码 + /// + public string TrackStatusCode { get; set; } + + /// + /// 轨迹发生地 + /// + public string TrackPlace { get; set; } = null!; + + /// + /// 轨迹发生时间 + /// + public DateTime TrackTime { get; set; } + + /// + /// 轨迹发生地时区 + /// + public float TrackTimeZone { get; set; } + + /// + /// 客户是否可见 + /// + public bool IsVisibility { get; set; } + + /// + /// 数据来源：1外部服务商创建，2内部系统创建，3人工创建 + /// + public int Source { get; set; } + + /// + /// 轨迹备注 + /// + public string? Remarks { get; set; } + + /// + /// 附件集合 + /// + public List? Files { get; set; } + + /// + /// 轨迹发生地（城市） + /// + public string? TrackPlaceCity { get; set; } + + /// + /// 轨迹发生地（省/州） + /// + public string? TrackPlaceProvince { get; set; } + + /// + /// 轨迹发生地（国家/地区） + /// + public string? TrackPlaceNation { get; set; } + + /// + /// 轨迹发生地邮编 + /// + public string? TrackPlacePostcode { get; set; } + + /// + /// 预计到达时间 + /// + public DateTime? EstimatedDeliveryDate { get; set; } +} diff --git a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index a5035d63..9e8a148b 100644 --- a/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs +++ b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs @@ -1,9 +1,12 @@ +using System.Text; +using BSI.Stack.Rabbit; using DotNetCore.CAP; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; +using SJZY.Track.Domain.Aggregates.TrackStatusAggregate; using SJZY.Track.Domain.Dtos.TrackContents; using SJZY.Track.Domain.Enums; using SJZY.Track.Domain.Queries; @@ -18,14 +21,21 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe private readonly IOrderRepository _orderRepository; private readonly IProvideQueries _provideQueries; private readonly ITrackDetailsEsRepository _trackDetailsEsRepository; + private readonly RabbitClientBus _rabbitClientBus; - public TrackUpsertEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher, - IProvideQueries provideQueries, ITrackDetailsEsRepository trackDetailsEsRepository) + public TrackUpsertEventHandler( + IOrderRepository orderRepository, + ICapPublisher capPublisher, + IProvideQueries provideQueries, + ITrackDetailsEsRepository trackDetailsEsRepository, + ITrackStatusRepository trackStatusRepository, + RabbitClientBus rabbitClientBus) { _orderRepository = orderRepository; _capPublisher = capPublisher; _provideQueries = provideQueries; _trackDetailsEsRepository = trackDetailsEsRepository; + _rabbitClientBus = rabbitClientBus; } [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)] @@ -35,13 +45,60 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe if (!CanDelete(@event)) return; var canNextOrderIds = new List(); - await HandleTrackAsync(@event, canNextOrderIds); + + var content = await HandleTrackAsync(@event, canNextOrderIds); + + // 处理广播事件 + PublicTrackTopic(@event, content); if (canNextOrderIds.Count == 0) return; // 根据订单关联关系异步向下刷（异步） await DownByOrderLink(@event, canNextOrderIds); } + /// + /// 发送轨迹广播 + /// + /// + /// + private void PublicTrackTopic(TrackUpsertEvent eto, TrackContentVO? content) + { + // var trackStatus = await _trackStatusRepository.GetFirstAsync(m => m.Code == content.TrackStatusCode); + if (content == null || !content.IsToMq) + return; + + var data = new TrackChangeEto() + { + EstimatedDeliveryDate = eto.EstimatedDeliveryDate, + TrackPlacePostcode = eto.TrackPlacePostcode, + TrackPlaceNation = eto.TrackPlaceNation, + TrackPlaceProvince = eto.TrackPlaceProvince, + TrackPlaceCity = eto.TrackPlaceCity, + Files = eto.Files, + Remarks = eto.Remarks, + Source = eto.Source, + IsVisibility = eto.IsVisibility, + TrackTimeZone = eto.TrackTimeZone, + TrackTime = eto.TrackTime, + TrackPlace = eto.TrackPlace, + TrackStatusCode = content.TrackStatusCode, + EnStatus = eto.EnStatus, + Content = eto.Content, + TrackCode = eto.TrackCode, + OperationMan = eto.OperationMan, + OrderIdentifiers = eto.OrderIdentifiers, + CustomId = eto.CustomId, + SubSystemId = eto.SubSystemId + }; + + if (eto.IsAdd) + data.Option = 1; + else + data.Option = 3; + + _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson())); + } + private async Task DownByOrderLink(TrackUpsertEvent @event, List canNextOrderIds) { // 根据订单关联关系异步向下刷（异步） @@ -126,7 +183,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe return true; } - private async Task HandleTrackAsync(TrackUpsertEvent @event, List canNextOrderIds) + private async Task HandleTrackAsync(TrackUpsertEvent @event, List canNextOrderIds) { var trackContent = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode); @@ -134,12 +191,12 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe { // 找不到轨迹内容，直接结束 LogHelper.Warn($"找不到轨迹内容：{@event}", true); - return; + return default; } var insertTracks = await UpsertTrackAsync(@event, canNextOrderIds, trackContent); - if (canNextOrderIds.Count == 0) return; + if (canNextOrderIds.Count == 0) return trackContent; if (insertTracks.Any()) { @@ -195,6 +252,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe } }; await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent); + + return trackContent; } private async Task> UpsertTrackAsync(TrackUpsertEvent @event, List canNextOrderIds, diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index 44486d40..e36b86ce 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs @@ -3,6 +3,7 @@ using DotNetCore.CAP; using DotNetCore.CAP.Internal; using Microsoft.Extensions.Configuration; using RabbitMQ.Client; +using Savorboard.CAP.InMemoryMessageQueue; using SJZY.Expand.ABP.Core.Localization; using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.PublicMethords; @@ -50,10 +51,7 @@ public class SjzyTrackApplicationModule : AbpModule }); //添加虚拟文件系统 - Configure(options => - { - options.FileSets.AddEmbedded("SJZY.Track.Application"); - }); + Configure(options => { options.FileSets.AddEmbedded("SJZY.Track.Application"); }); //添加多语言 var languageConfig = configuration.GetSection("Language"); @@ -82,45 +80,59 @@ public class SjzyTrackApplicationModule : AbpModule //CAP context.Services.AddSingleton(); context.Services.AddCapTableNameService(); - context.Services.AddCap(options => + if (!System.Diagnostics.Debugger.IsAttached) { - options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!); - options.UseRabbitMQ(configure => + context.Services.AddCap(options => { - configure.HostName = configuration["RabbitMQ:HostName"]!; - configure.UserName = configuration["RabbitMQ:UserName"]!; - configure.Password = configuration["RabbitMQ:Password"]!; - configure.Port = configuration.GetValue("RabbitMQ:Port"); - configure.BasicQosOptions = - new RabbitMQOptions.BasicQos(configuration.GetValue("CAP:PrefetchCount")); - }); - options.FailedRetryInterval = configuration.GetValue("CAP:FailedRetryInterval"); - options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount"); - options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter"); - options.FallbackWindowLookbackSeconds = - configuration.GetValue("CAP:FallbackWindowLookbackSeconds"); - - options.UseDashboard(dashboardOptions => - { - dashboardOptions.PathBase = System.Diagnostics.Debugger.IsAttached - ? "" - : configuration["Swagger:JsonUrlPrefix"]!; - }); - options.FailedThresholdCallback = faileCalback => - { - try + options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!); + options.UseRabbitMQ(configure => { - LogHelper.Error($@" + configure.HostName = configuration["RabbitMQ:HostName"]!; + configure.UserName = configuration["RabbitMQ:UserName"]!; + configure.Password = configuration["RabbitMQ:Password"]!; + configure.Port = configuration.GetValue("RabbitMQ:Port"); + configure.BasicQosOptions = + new RabbitMQOptions.BasicQos(configuration.GetValue("CAP:PrefetchCount")); + }); + options.FailedRetryInterval = configuration.GetValue("CAP:FailedRetryInterval"); + options.FailedRetryCount = configuration.GetValue("CAP:FailedRetryCount"); + options.SucceedMessageExpiredAfter = configuration.GetValue("CAP:SucceedMessageExpiredAfter"); + options.FallbackWindowLookbackSeconds = + configuration.GetValue("CAP:FallbackWindowLookbackSeconds"); + + options.UseDashboard(dashboardOptions => + { + dashboardOptions.PathBase = System.Diagnostics.Debugger.IsAttached + ? "" + : configuration["Swagger:JsonUrlPrefix"]!; + }); + options.FailedThresholdCallback = faileCalback => + { + try + { + LogHelper.Error($@" |-CAP失败回调事件： |-处理类型：{(faileCalback.MessageType == DotNetCore.CAP.Messages.MessageType.Subscribe ? "消费消息达到最大重试次数，需要开发人员介入" : "生产消息推送失败")} |-头部信息：{faileCalback.Message.Headers.ToJson()}"); - } - catch (Exception ex) - { - LogHelper.Error("【严重】CAP失败回调事件发生未处理异常" + ex); - } - }; - }).AddSubscribeFilter(); + } + catch (Exception ex) + { + LogHelper.Error("【严重】CAP失败回调事件发生未处理异常" + ex); + } + }; + }).AddSubscribeFilter(); + } + else + { + //是本地调试环境，用内存存储和内存队列 + context.Services.AddCap(capOptions => + { + capOptions.UseInMemoryStorage(); + capOptions.UseInMemoryMessageQueue(); + capOptions.UseDashboard(); + }).AddSubscribeFilter(); + Console.WriteLine("-- 当前为VS本地环境，用内存存储和内存队列"); + } context.Services.AddRabbitBoot(options => @@ -142,6 +154,17 @@ public class SjzyTrackApplicationModule : AbpModule o.QueueBind(GlobalConst.TrackRefreshSubscribe, GlobalConst.InterfaceRelayExchange, GlobalConst.TrackRefreshSubscribe); } + }, + new() + { + //广播轨迹 + Queue = GlobalConst.TrackRefreshSubscribe, + FetchCount = configuration.GetValue("TrackApp:TrackRefreshFetchCount"), + FailedRequeue = false, + Declaration = o => + { + o.ExchangeDeclare(GlobalConst.TrackChangeExchange, ExchangeType.Topic, true); + } } }; }, typeof(SjzyTrackApplicationModule).Assembly); diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index d517bfc8..e7901efa 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -51,4 +51,11 @@ public partial class GlobalConst /// 预警计算任务 /// public const string TrackWarningCalcSubscribe = "SJZY.Track.TrackWarningCalc"; + + /// + /// 轨迹变更通知交换机 + /// + public const string TrackChangeExchange = "sjzy.track.on_track_change_ex"; + + } diff --git a/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs b/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs index f9f6da60..f084b0a8 100644 --- a/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs +++ b/src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs @@ -51,4 +51,9 @@ public class TrackContentVO /// 是否允许手动创建 /// public bool IsAllowManuallyCreate { get; set; } + + /// + /// 是否广播 + /// + public bool IsToMq { get; set; } } diff --git a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs index 25d2c0d9..710dd407 100644 --- a/src/SJZY.Track.Repository/Queries/ProvideQueries.cs +++ b/src/SJZY.Track.Repository/Queries/ProvideQueries.cs @@ -93,7 +93,8 @@ public class ProvideQueries : ISugarRepository, IProvideQueries, ITransientDepen IsCompleted = y.IsCompleted, IsRepeatable = x.IsRepeatable, IsAllowManuallyUpdate = x.IsAllowManuallyUpdate, - IsAllowManuallyCreate = x.IsAllowManuallyCreate + IsAllowManuallyCreate = x.IsAllowManuallyCreate, + IsToMq = y.IsToMq, }) .FirstAsync(); }