update: 优化轨迹相关广播,抽离广播为BroadcastingService，涉及轨迹变更，轨迹推送完成。

update-eto:
 - rename: TrackPushEto -> TrackPushCompletedEvent

src/SJZY.Track.Application.Contract/Events/TrackPushCompletedEvent.cs

@@ -0,0 +1,82 @@
+namespace SJZY.Track.Application.Contract.Events;
+
+/// <summary>
+/// 轨迹推送广播模型
+/// </summary>
+public class TrackPushCompletedEvent
+{
+    /// <summary>
+    /// 是否为手动推送
+    /// </summary>
+    public bool IsManualPush { get; set; }
+
+    /// <summary>
+    /// 推送时间
+    /// </summary>
+    public DateTime PushTime { get; set; }
+
+    /// <summary>
+    /// 操作人
+    /// </summary>
+    public string? OperationMan { get; set; }
+
+    /// <summary>
+    /// 订单号
+    /// </summary>
+    public string OrderNo { get; set; } = null!;
+
+    /// <summary>
+    /// 接口通请求Id
+    /// </summary>
+    public Guid RequestId { get; set; }
+
+    /// <summary>
+    /// 信息
+    /// </summary>
+    public string Message { get; set; } = null!;
+
+    /// <summary>
+    /// 操作类型
+    /// </summary>
+    public string OptionType { get; set; } = null!;
+
+    /// <summary>
+    /// 轨迹状态
+    /// </summary>
+    public string TrackStatus { get; set; } = null!;
+
+    /// <summary>
+    /// 轨迹内容代码
+    /// </summary>
+    public string TrackCode { get; set; } = null!;
+
+    /// <summary>
+    /// 轨迹内容
+    /// </summary>
+    public string TrackContent { get; set; } = null!;
+
+    /// <summary>
+    /// 轨迹发生地
+    /// </summary>
+    public string TrackPlace { get; set; } = null!;
+
+    /// <summary>
+    /// 轨迹发生时间
+    /// </summary>
+    public DateTime TrackTime { get; set; }
+
+    /// <summary>
+    /// 轨迹发生时区
+    /// </summary>
+    public float TrackTimeZone { get; set; }
+
+    /// <summary>
+    /// 是否成功
+    /// </summary>
+    public bool IsSuccess { get; set; }
+
+    /// <summary>
+    /// 推送内容（json）
+    /// </summary>
+    public string PushContent { get; set; } = null!;
+}

src/SJZY.Track.Application.Contract/Events/TrackPushEvent.cs

@@ -46,6 +46,11 @@ public class TrackPushEvent
     /// 是否强制推送
     /// </summary>
     public bool ForcedPush { get; set; }
+
+    /// <summary>
+    /// 是否发送到mq
+    /// </summary>
+    public bool IsToMq { get; set; }
 }
 
 public struct SystemPushStruct

src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs

@@ -1,11 +1,8 @@
-using System.Text.Json;
 using DotNetCore.CAP;
 using Microsoft.Extensions.Options;
 using Newtonsoft.Json;
-using Newtonsoft.Json.Linq;
 using SJZY.Expand.ABP.Core.PublicMethords;
 using SJZY.InterfaceRelay.Application.Contracts;
-using SJZY.OTel;
 using SJZY.Track.Application.Contract.Dtos.TrackPush;
 using SJZY.Track.Application.Contract.Events;
 using SJZY.Track.Application.Contract.Services;
@@ -127,6 +124,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
                     TrackTime = @event.SystemPushInfo.TrackTime,
                     TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
                     CreateUser = GlobalConst.SystemUser,
+                    SubSystemCode = @event.SubSystemCode,
+                    IsToMq = @event.IsToMq
                 };
                 var item = await BuiltPushItem(config, log, @event.SystemPushInfo.Extra);
                 item.Log = JsonConvert.SerializeObject(log);
@@ -173,6 +172,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
                     TrackTime = track.TrackTime,
                     TrackTimeZone = track.TrackTimeZone,
                     CreateUser = @event.OperationMan,
+                    SubSystemCode = @event.SubSystemCode,
+                    IsToMq = @event.IsToMq
                 };
                 var item = await BuiltPushItem(config, log, track.Extra);
                 item.Log = JsonConvert.SerializeObject(log);
@@ -182,20 +183,6 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
 
         if (items.Count == 0) return;
 
-        //获取参数结构名称
-        // var paramNames = await GetExtraParamNamesAsync(@event.SubSystemCode);
-        //
-        // //如果为系统推送，则从事件中取出轨迹扩展内容填充
-        // foreach (var item in items)
-        // {
-        //     //根据事件中轨迹扩展内容填充
-        //     if (isSystemPush)
-        //         item.Data.ForEach(dict => FillExtraParam(@event.SystemPushInfo.Extra, dict, paramNames));
-        //     //根据查出的轨迹中扩展内容填充
-        //     else
-        //         item.Data.ForEach(dict => FillExtraParam(trackPushInputDict[item].Extra, dict, paramNames));
-        // }
-
         using (Track2InterfaceRelaySubSystemContext.Use(
                    new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode)))
         {
@@ -235,6 +222,8 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
                 TrackTime = @event.SystemPushInfo.TrackTime,
                 TrackTimeZone = @event.SystemPushInfo.TrackTimeZone,
                 CreateUser = GlobalConst.SystemUser,
+                SubSystemCode = @event.SubSystemCode,
+                IsToMq = @event.IsToMq
             };
             var item = BuiltUpdateItem(config, log, @event.SystemPushInfo.Extra);
             item.Log = JsonConvert.SerializeObject(log);
@@ -243,13 +232,6 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
 
         if (items.Count == 0) return;
 
-        // //获取参数结构名称
-        // var paramNames = await GetExtraParamNamesAsync(@event.SubSystemCode);
-        //
-        // //根据事件中轨迹扩展内容填充请求
-        // foreach (var item in items)
-        //     item.Data.ForEach(dict => FillExtraParam(@event.SystemPushInfo.Extra, dict, paramNames));
-
         using (Track2InterfaceRelaySubSystemContext.Use(
                    new Track2InterfaceRelaySubSystemContext(@event.SubSystemCode)))
         {
@@ -384,114 +366,9 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
     {
         return obj == null ? string.Empty : System.Text.Json.JsonSerializer.Serialize(obj);
     }
+
     private T? DeserializeForTextJson<T>(string? json)
     {
         return string.IsNullOrWhiteSpace(json) ? default : System.Text.Json.JsonSerializer.Deserialize<T>(json);
     }
-
-
-    /*
-     *  review by xiaolipro at 2024-12-24
-     *  todo：无用的方法，可以删除 @王云龙
-     */
-
-    /// <summary>
-    /// 获取来源系统接口的参数名
-    /// </summary>
-    /// <param name="subSystemCode"></param>
-    /// <returns></returns>
-    private async Task<List<string>> GetExtraParamNamesAsync(string subSystemCode)
-    {
-        using var span = ActivityHelper.CreateSpanActivity("开始请求接口通，接口结构");
-        using (Track2InterfaceRelaySubSystemContext.Use(
-                   new Track2InterfaceRelaySubSystemContext(subSystemCode)))
-        {
-            var res = await _httpClientProxy.Service.GetSystemInterfaceByIdAsync(_options.CurrentValue.PushTrackInterfaceCode, _options.CurrentValue.PushTrackInterfaceId);
-            if (res == null) return new List<string>();
-            span.SetTag("request-GetSystemInterfaceByIdAsync", $"PushTrackInterfaceCode[{subSystemCode}]," +
-                                                               $"PushTrackInterfaceCode:[{_options.CurrentValue.PushTrackInterfaceCode}]," +
-                                                               $"PushTrackInterfaceId:[{_options.CurrentValue.PushTrackInterfaceId}]," +
-                                                               $"PushTrackItemParamAliasPrefix:[{_options.CurrentValue.PushTrackItemParamAliasPrefix}]");
-            span.SetTag("response-GetSystemInterfaceByIdAsync", res.ToJson());
-            return res.InParameterList
-                .Where(x => x.Alias.StartsWith(_options.CurrentValue.PushTrackItemParamAliasPrefix))
-                .Select(x => x.Name)
-                .ToList();
-        }
-    }
-
-    /// <summary>
-    /// 填充来源于轨迹扩展字段的参数
-    /// </summary>
-    /// <param name="trackExtraJson">轨迹明细扩展字段Json</param>
-    /// <param name="dict">要扩展的参数字典</param>
-    /// <param name="paramNames">来源扩展字段的参数名</param>
-    /// <returns></returns>
-    private void FillExtraParam(string? trackExtraJson, Dictionary<string, object>? dict, List<string>? paramNames)
-    {
-        if (paramNames == null || paramNames.Count == 0) return;
-        if (dict == null) return;
-
-        if (string.IsNullOrWhiteSpace(trackExtraJson)) return;
-
-        dict.TryAdd("extra",JObject.FromObject(trackExtraJson));
-        var trackExtra = ExtractTrackExtraDict(trackExtraJson);
-
-        if (trackExtra.Count == 0) return;
-        foreach (var item in
-                 trackExtra.Where(x => x.Value != null && paramNames.Contains(x.Key)))
-        {
-            dict.TryAdd(item.Key, item.Value);
-        }
-    }
-
-    private Dictionary<string, object> ExtractTrackExtraDict(string trackExtraStr)
-    {
-        var dict = new Dictionary<string, object>();
-        var trackExtra = JsonConvert.DeserializeObject<Dictionary<string, object>>(trackExtraStr);
-        if (trackExtra == null) return dict;
-        foreach (var pair in trackExtra)
-        {
-            if (pair.Value is JsonElement jEl)
-            {
-                object? objVal = null;
-                switch (jEl.ValueKind)
-                {
-                    case JsonValueKind.String:
-                        objVal = jEl.GetString();
-                        // 处理字符串值
-                        break;
-                    case JsonValueKind.Number:
-                        objVal = jEl.GetDouble();
-                        // 处理数值
-                        break;
-                    case JsonValueKind.True:
-                    case JsonValueKind.False:
-                        objVal = jEl.GetBoolean();
-                        // 处理布尔值
-                        break;
-                    case JsonValueKind.Object:
-                        // 处理对象
-                        break;
-                    case JsonValueKind.Array:
-                        // 处理数组
-                        break;
-                    case JsonValueKind.Null:
-                        // 处理空值
-                        break;
-                    default:
-                        // 处理未知类型
-                        break;
-                }
-
-                if (objVal == null) continue;
-                dict.TryAdd(pair.Key, objVal);
-                continue;
-            }
-
-            dict.TryAdd(pair.Key, pair.Value);
-        }
-
-        return dict;
-    }
 }

src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs

@@ -1,9 +1,8 @@
-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.Application.Services.Broadcasting;
 using SJZY.Track.Domain.Aggregates.OrderAggregate;
 using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
 using SJZY.Track.Domain.Dtos.TrackContents;
@@ -20,7 +19,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
     private readonly IOrderRepository _orderRepository;
     private readonly IProvideQueries _provideQueries;
     private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
-    private readonly RabbitClientBus _rabbitClientBus;
+    private readonly BroadcastingService _broadcastingService;
     private readonly IOrderOperationRecordEsRepository _orderOperationRecordEsRepository;
 
     public TrackUpsertEventHandler(
@@ -28,15 +27,15 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
         ICapPublisher capPublisher,
         IProvideQueries provideQueries,
         ITrackDetailsEsRepository trackDetailsEsRepository,
-        RabbitClientBus rabbitClientBus,
+        IOrderOperationRecordEsRepository orderOperationRecordEsRepository,
-        IOrderOperationRecordEsRepository orderOperationRecordEsRepository)
+        BroadcastingService broadcastingService)
     {
         _orderRepository = orderRepository;
         _capPublisher = capPublisher;
         _provideQueries = provideQueries;
         _trackDetailsEsRepository = trackDetailsEsRepository;
-        _rabbitClientBus = rabbitClientBus;
         _orderOperationRecordEsRepository = orderOperationRecordEsRepository;
+        _broadcastingService = broadcastingService;
     }
 
     [CapSubscribe(GlobalConst.TrackUpsertSubscribe, Group = GlobalConst.TrackUpsertSubscribe)]
@@ -49,62 +48,15 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
 
         var content = await HandleTrackAsync(@event, canNextOrderIds);
 
-        // todo: 广播提取到一个服务里去，异步处理，不影响主流程
         // 处理广播事件
-        PublicTrackTopic(@event, content);
+        if (content is { IsToMq: true })
+            await _broadcastingService.TrackChangePublicTopicAsync(@event, content);
 
         if (canNextOrderIds.Count == 0) return;
         // 根据订单关联关系异步向下刷（异步）
         await DownByOrderLink(@event, canNextOrderIds);
     }
 
-    /// <summary>
-    /// 发送轨迹广播
-    /// </summary>
-    /// <param name="eto"></param>
-    /// <param name="content"></param>
-    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,
-            TrackPlaceNationId = eto.TrackPlaceNationId,
-            TrackPlaceProvince = eto.TrackPlaceProvince,
-            TrackPlaceProvinceId = eto.TrackPlaceProvinceId,
-            TrackPlaceCity = eto.TrackPlaceCity,
-            TrackPlaceCityId = eto.TrackPlaceCityId,
-            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,
-            Extra = eto.Extra
-        };
-
-        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<int> canNextOrderIds)
     {
         // 根据订单关联关系异步向下刷（异步）
@@ -268,7 +220,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
                 TrackTimeZone = @event.TrackTimeZone,
                 Extra = @event.Extra
             },
-            ForcedPush = @event.ForcedPush
+            ForcedPush = @event.ForcedPush,
+            IsToMq = trackContent.IsToMq
         };
         await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
 

src/SJZY.Track.Application/Rabbit/TrackPushCompletedEventHandler.cs

@@ -4,17 +4,12 @@ using BSI.Stack.Rabbit.Handler;
 using Newtonsoft.Json;
 using RabbitMQ.Client.Events;
 using SJZY.Expand.ABP.Core.Logger;
-using SJZY.Expand.ABP.Core.PublicMethords;
 using SJZY.InterfaceRelay.Application.Contracts;
-using SJZY.Track.Application.Contract.Dtos.ProvideServices;
 using SJZY.Track.Application.Contract.Dtos.TrackPush;
-using SJZY.Track.Application.Contract.Events;
 using SJZY.Track.Application.Helpers;
+using SJZY.Track.Application.Services.Broadcasting;
 using SJZY.Track.Domain.Aggregates.OrderAggregate;
-using SJZY.Track.Domain.Aggregates.TrackContentAggregate;
-using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
 using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
-using SJZY.Track.Domain.Aggregates.TrackStatusAggregate;
 using SJZY.Track.Domain.Enums;
 using Volo.Abp.DependencyInjection;
 
@@ -24,24 +19,15 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
 {
     private readonly IOrderRepository _orderRepository;
     private readonly ITrackPushRecordEsRepository _trackPushRecordEsRepository;
-    private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
+    private readonly BroadcastingService _broadcastingService;
-    private readonly ITrackStatusRepository _trackStatusRepository;
-    private readonly ITrackContentRepository _trackContentRepository;
-    private readonly RabbitClientBus _rabbitClientBus;
 
     public TrackPushCompletedEventHandler(IOrderRepository orderRepository,
         ITrackPushRecordEsRepository trackPushRecordEsRepository,
-        ITrackDetailsEsRepository trackDetailsEsRepository,
+        BroadcastingService broadcastingService)
-        ITrackStatusRepository trackStatusRepository,
-        ITrackContentRepository trackContentRepository,
-        RabbitClientBus rabbitClientBus)
     {
         _orderRepository = orderRepository;
         _trackPushRecordEsRepository = trackPushRecordEsRepository;
-        _trackDetailsEsRepository = trackDetailsEsRepository;
+        _broadcastingService = broadcastingService;
-        _trackStatusRepository = trackStatusRepository;
-        _trackContentRepository = trackContentRepository;
-        _rabbitClientBus = rabbitClientBus;
     }
 
     public ConsumeOptions Options()
@@ -71,7 +57,7 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
         }
         else if (optionType == OptionType.修改.ToString())
         {
-            logStr = JsonConvert.DeserializeObject<List<TrackUpdateInput>>(@event.RequestParams)!.First().Log;;
+            logStr = JsonConvert.DeserializeObject<List<TrackUpdateInput>>(@event.RequestParams)!.First().Log;
         }
 
         if (logStr.IsNullOrEmpty()) throw new InvalidOperationException();
@@ -82,16 +68,9 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
         log.Message = @event.Message;
         log.RequestId = @event.RequestId;
 
-        // todo：这里不需要查es，log里面有
-        //查出推送轨迹，写入推送广播
-        var tracks = await _trackDetailsEsRepository
-            .GetListAsync(new[] { log.OrderId }, new List<string>
-            {
-                log.TrackCode
-            });
-
         // 推送广播
-        await PublicTrackPushTopicAsync(tracks.First(), log, !log.IsSuccess, log.Message);
+        if (log.IsToMq)
+            await _broadcastingService.TrackPushCompletedPublicTopicAsync(log);
 
         // 往es 写推送日志
         await _trackPushRecordEsRepository.AddAsync(log);
@@ -140,72 +119,4 @@ public class TrackPushCompletedEventHandler : ITransientDependency, IRabbitHandl
                 });
         }
     }
-
-    /// <summary>
-    /// 轨迹推送广播
-    /// </summary>
-    private async Task PublicTrackPushTopicAsync(TrackDetailsIndex track, TrackPushRecordIndex pushRecord, bool isFailure = false, string? errorMsg = null)
-    {
-        // view by xiaolipro at 2024-12-24
-        // todo：热路径下进行数据库查询影响性能，建议减少或缓存，或异步处理@王云龙
-        var statusList = await _trackStatusRepository.GetListAsync(m => m.SubSystemId == track.SubSystemId);
-        if (statusList == null)
-            return;
-
-        var contentList = await _trackContentRepository.GetTrackContentBySyetemIdAsync(track.SubSystemId);
-        if (contentList == null)
-            return;
-
-        var status = statusList.FirstOrDefault(m => m.Code == track.TrackStatusCode);
-        if (status == null || !status.IsToMq)
-            return;
-
-        var content = contentList.FirstOrDefault(m => m.TrackContentCode == track.TrackCode);
-
-        var data = new TrackPushEto
-        {
-            Option = pushRecord.OptionType == OptionType.新增.ToString() ? OptionType.新增 : OptionType.修改,
-            IsManualPush = !string.Equals(pushRecord.CreateUser, GlobalConst.SystemUser), //操作人不为系统用户，则认为是仍推送
-            IsFailure = isFailure,
-            FailureReason = errorMsg,
-            PushTime = DateTime.Now,
-            OperationMan = pushRecord.CreateUser,
-            EstimatedDeliveryDate = track.EstimatedDeliveryDate,
-            TrackPlacePostcode = track.TrackPlacePostcode,
-            TrackPlaceNation = track.TrackPlaceNation,
-            TrackPlaceNationId = track.TrackPlaceNationId,
-            TrackPlaceProvince = track.TrackPlaceProvince,
-            TrackPlaceProvinceId = track.TrackPlaceProvinceId,
-            TrackPlaceCity = track.TrackPlaceCity,
-            TrackPlaceCityId = track.TrackPlaceCityId,
-            Files = track.Files?.Select(x => new FileItemDto
-            {
-                OssKey = x.OssKey,
-                FileName = x.FileName
-            }).ToList(),
-            Remarks = track.Remarks,
-            Source = track.Source,
-            IsVisibility = track.IsVisibility,
-            TrackTimeZone = track.TrackTimeZone,
-            TrackTime = track.TrackTime,
-            TrackPlace = track.TrackPlace,
-            TrackStatusCode = track.TrackStatusCode,
-            EnStatus = track.TrackEnContent,
-            Content = track.TrackContent,
-            TrackCode = track.TrackCode,
-            SubSystemId = track.SubSystemId,
-            OrderNo = track.OrderNo,
-            OrderIdentifiers = new OrderIdentifier[]
-            {
-                new OrderIdentifier
-                {
-                    OrderId = track.OrderId,
-                    OrderNo = track.OrderNo,
-                    OrderType = track.OrderType
-                }
-            },
-            Extra = track.Extra
-        };
-        _rabbitClientBus.Publish(GlobalConst.TrackPushExchange, $"TrackPush.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson()));
-    }
 }

src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs

@@ -0,0 +1,103 @@
+using System.Text;
+using BSI.Stack.Rabbit;
+using SJZY.Expand.ABP.Core.PublicMethords;
+using SJZY.Track.Application.Contract.Events;
+using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
+using SJZY.Track.Domain.Dtos.TrackContents;
+using Volo.Abp.DependencyInjection;
+
+namespace SJZY.Track.Application.Services.Broadcasting;
+
+/// <summary>
+/// 广播服务
+/// </summary>
+public class BroadcastingService : ITransientDependency
+{
+    private readonly RabbitClientBus _rabbitClientBus;
+
+    public BroadcastingService(RabbitClientBus rabbitClientBus)
+    {
+        _rabbitClientBus = rabbitClientBus;
+    }
+
+    /// <summary>
+    /// 轨迹变更广播
+    /// </summary>
+    /// <param name="eto"></param>
+    /// <param name="content"></param>
+    /// <returns></returns>
+    public Task TrackChangePublicTopicAsync(TrackUpsertEvent eto, TrackContentVO content)
+    {
+        return Task.Run(() =>
+        {
+            var data = new TrackChangeEto
+            {
+                EstimatedDeliveryDate = eto.EstimatedDeliveryDate,
+                TrackPlacePostcode = eto.TrackPlacePostcode,
+                TrackPlaceNation = eto.TrackPlaceNation,
+                TrackPlaceNationId = eto.TrackPlaceNationId,
+                TrackPlaceProvince = eto.TrackPlaceProvince,
+                TrackPlaceProvinceId = eto.TrackPlaceProvinceId,
+                TrackPlaceCity = eto.TrackPlaceCity,
+                TrackPlaceCityId = eto.TrackPlaceCityId,
+                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,
+                Extra = eto.Extra
+            };
+
+            if (eto.IsAdd)
+                data.Option = 1;
+            else
+                data.Option = 3;
+
+            _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{content.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson()));
+            return Task.CompletedTask;
+        });
+    }
+
+    /// <summary>
+    /// 轨迹推送完成广播
+    /// </summary>
+    /// <param name="log"></param>
+    /// <returns></returns>
+    public Task TrackPushCompletedPublicTopicAsync(TrackPushRecordIndex log)
+    {
+        return Task.Run(() =>
+        {
+            var data = new TrackPushCompletedEvent
+            {
+                RequestId = log.RequestId,
+                IsSuccess = log.IsSuccess,
+                Message = log.Message,
+                OperationMan = log.CreateUser,
+                PushTime = DateTime.Now,
+                OptionType = log.OptionType,
+                IsManualPush = !string.Equals(log.CreateUser, GlobalConst.SystemUser), //操作人不为系统用户，则认为是仍推送
+                PushContent = log.PushContent,
+                OrderNo = log.OrderNo,
+                TrackStatus = log.TrackStatus,
+                TrackContent = log.TrackContent,
+                TrackTimeZone = log.TrackTimeZone,
+                TrackTime = log.TrackTime,
+                TrackPlace = log.TrackPlace,
+                TrackCode = log.TrackCode
+            };
+
+            _rabbitClientBus.Publish(GlobalConst.TrackPushExchange, $"TrackPush.{log.SubSystemCode}", Encoding.UTF8.GetBytes(data.ToJson()));
+            return Task.CompletedTask;
+        });
+    }
+}

src/SJZY.Track.Domain/Aggregates/TrackPushAggregate/TrackPushRecordIndex.cs

@@ -82,4 +82,14 @@ public class TrackPushRecordIndex : IdBaseIndex
 
     /// <summary>创建时间</summary>
     public DateTime CreateTime { get; set; }
+
+    /// <summary>
+    /// 子系统代码
+    /// </summary>
+    public string SubSystemCode { get; set; }
+
+    /// <summary>
+    /// 是否广播
+    /// </summary>
+    public bool IsToMq { get; set; }
 }