Merge branch 'refs/heads/S03-轨迹中台优化项' into dev

src/SJZY.Track.Api/appsettings.json

@@ -5,7 +5,7 @@
     "Apollo": {
       "AppId": "SJZY.Track.Configuration",
       "Env": "DEV",
-      "Namespace": "Dev",
+      "Namespace": "Test",
       "MetaServer": "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080",
       "ConfigServer": [ "http://apollo-service-test-apollo-configservice.apollo.svc.cluster.local:8080" ]
     }

src/SJZY.Track.Application/CAP/CAPFilter.cs

@@ -0,0 +1,97 @@
+using System.Security.Cryptography;
+using System.Text;
+using DotNetCore.CAP.Filter;
+using SJZY.Expand.ABP.Core;
+using SJZY.Expand.ABP.Core.Logger;
+using SJZY.Expand.ABP.Core.PublicMethords;
+using SJZY.Track.Domain.Consts;
+
+namespace SJZY.Track.Application.CAP;
+
+public class CapFilter : SubscribeFilter
+{
+    const string ErrorCode = "Error599";
+
+    /// <summary>
+    /// 执行前
+    /// </summary>
+    /// <param name="context"></param>
+    /// <returns></returns>
+    public override async Task OnSubscribeExecutingAsync(ExecutingContext context)
+    {
+        // 轨迹新增、更新要做幂等
+        if (!context.DeliverMessage.Headers["cap-msg-name"]
+                .IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe))
+        {
+            return;
+        }
+
+        var argumentJsonStr = context.Arguments.ToJson();
+
+        if (!string.IsNullOrEmpty(argumentJsonStr))
+        {
+            context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash);
+            if (hash.IsNullOrEmpty()) // 计算参数hash
+            {
+                hash = GetHash(argumentJsonStr);
+                context.DeliverMessage.Headers["sjzy-arguments-hash"] = hash;
+            }
+
+            if (await RedisHelper.Client.ExistsAsync($"CapExecuteFilter:{hash}"))
+            {
+                throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}");
+            }
+        }
+    }
+
+    public override async Task OnSubscribeExecutedAsync(ExecutedContext context)
+    {
+        context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash);
+        if (!hash.IsNullOrEmpty())
+        {
+            await RedisHelper.Client.SetAsync($"CapExecuteFilter:{hash}", 1, TimeSpan.FromMinutes(10).Seconds);
+        }
+    }
+
+    /// <summary>
+    /// 消费异常事件
+    /// </summary>
+    /// <param name="context"></param>
+    /// <returns></returns>
+    public override Task OnSubscribeExceptionAsync(ExceptionContext context)
+    {
+        context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
+
+        if (context.Exception is BusinessException ex && ex.Code == ErrorCode)
+        {
+            context.ExceptionHandled = true; // 幂等不做重试
+            LogHelper.Warn(ex.ToString());
+        }
+        else
+        {
+            // 发飞书，异常继续上抛，进入重试机制
+            // tips：重试有冷却期（默认4分钟）
+            LogHelper.Error($"cap-msg-id：{messageId}\n{context.Exception}");
+        }
+
+        return Task.CompletedTask;
+    }
+
+
+    /// <summary>获取字符串的hash值</summary>
+    /// <param name="str"></param>
+    /// <returns></returns>
+    public static string GetHash(string str)
+    {
+        // 将字符串转换为字节
+        byte[] data = Encoding.UTF8.GetBytes(str);
+
+        // 创建SHA256实例
+        using SHA256 sha256 = SHA256.Create();
+        // 计算哈希值
+        byte[] hash = sha256.ComputeHash(data);
+
+        // 将哈希值转换为16进制字符串
+        return BitConverter.ToString(hash);
+    }
+}

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

@@ -5,7 +5,7 @@ using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
 using SJZY.Track.Domain.Consts;
 using Volo.Abp.DependencyInjection;
 
-namespace SJZY.Track.Application.EventHandlers;
+namespace SJZY.Track.Application.CAP.EventHandlers;
 
 public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICapSubscribe
 {
@@ -36,6 +36,7 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa
 
         if (orders.Count == 0) return;
 
+        var updateOrders = new List<OrderEntity>();
         foreach (var order in orders)
         {
             if (order.LatestTrackContent != @event.TrackContent)
@@ -58,9 +59,10 @@ public class TrackChangeAfterUpdateOrderEventHandler : ITransientDependency, ICa
             order.LatestTrackPlace = @event.TrackPlace;
             order.LatestTrackTime = @event.TrackTime;
             order.LatestTrackTimeZone = @event.TrackTimeZone;
+            updateOrders.Add(order);
         }
 
-        await _orderRepository.UpdateRangeAsync(orders, x => new
+        await _orderRepository.UpdateRangeAsync(updateOrders, x => new
         {
             x.IsCompleted, x.LatestTrackContent, x.LatestTrackEnContent, x.LatestTrackPlace, x.LatestTrackTime,
             x.LatestTrackTimeZone

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

@@ -8,7 +8,7 @@ using SJZY.Track.Domain.Consts;
 using SJZY.Track.Domain.Queries;
 using Volo.Abp.DependencyInjection;
 
-namespace SJZY.Track.Application.EventHandlers;
+namespace SJZY.Track.Application.CAP.EventHandlers;
 
 public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
 {
@@ -109,13 +109,14 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
             // 删除全部轨迹导致的重置行为
             if (track == default)
             {
-                if (order.LatestTrackTime == null) return;
+                if (order.LatestTrackTime == null) continue;
 
                 // tips：订单是支持从已完成态到未完成态的
                 order.ResetTrackInfo();
             }
             else
             {
+                if (order.LatestTrackContent == track.TrackContent) continue;
                 // 删除引发的回退更新
                 order.LatestTrackContent = track.TrackContent;
                 order.LatestTrackEnContent = track.TrackEnContent;

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

@@ -4,7 +4,7 @@ using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
 using SJZY.Track.Domain.Consts;
 using Volo.Abp.DependencyInjection;
 
-namespace SJZY.Track.Application.EventHandlers;
+namespace SJZY.Track.Application.CAP.EventHandlers;
 
 public class TrackInsertForNotMappedEventHandler : ITransientDependency, ICapSubscribe
 {

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

@@ -9,7 +9,7 @@ using SJZY.Track.Domain.Consts;
 using SJZY.Track.Domain.Enums;
 using Volo.Abp.DependencyInjection;
 
-namespace SJZY.Track.Application.EventHandlers;
+namespace SJZY.Track.Application.CAP.EventHandlers;
 
 public class TrackPushCompletedEventHandler : ITransientDependency, ICapSubscribe
 {
@@ -65,8 +65,14 @@ public class TrackPushCompletedEventHandler : ITransientDependency, ICapSubscrib
 
     private async Task TrackPushAfterUpdateOrderAsync(TrackPushRecordIndex log)
     {
-        var order = new OrderEntity();
+        // tips：只有新增会影响推送进度？
-        order.Id = log.OrderId;
+        if (log.OptionType != OptionType.新增.ToString()) return;
+
+        var order = new OrderEntity
+        {
+            Id = log.OrderId
+        };
+
         /*
          *  异常：最近一条推送是失败
          */
@@ -74,27 +80,22 @@ public class TrackPushCompletedEventHandler : ITransientDependency, ICapSubscrib
         {
             order.QuestionType = (int)TrackProblemType.Error;
             order.LatestPushError = log.Message;
+            await _orderRepository.UpdateColumnAsync(order, x => new
+            {
+                x.QuestionType, x.LatestPushError
+            });
         }
         else
         {
+            // 还原成正常态
             order.QuestionType = (int)TrackProblemType.None;
             order.LatestPushError = null;
-            if (log.OptionType == OptionType.新增.ToString())
+            order.LatestPushTrackCode = log.TrackCode;
+            order.LatestPushTime = log.CreateTime;
+            await _orderRepository.UpdateColumnAsync(order, x => new
             {
-                // tips：只有新增会影响推送进度？
+                x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError
-                order.LatestPushTrackCode = log.TrackCode;
+            });
-                order.LatestPushTime = DateTime.Now;
-                await _orderRepository.UpdateColumnAsync(order, x => new
-                {
-                    x.LatestPushTrackCode, x.LatestPushTime, x.QuestionType, x.LatestPushError
-                });
-                return;
-            }
         }
-
-        await _orderRepository.UpdateColumnAsync(order, x => new
-        {
-            x.QuestionType, x.LatestPushError
-        });
     }
 }

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

@@ -14,7 +14,7 @@ using SJZY.Track.Domain.Enums;
 using Volo.Abp.DependencyInjection;
 using Volo.Abp.Http.Client.DynamicProxying;
 
-namespace SJZY.Track.Application.EventHandlers;
+namespace SJZY.Track.Application.CAP.EventHandlers;
 
 public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
 {
@@ -38,15 +38,15 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
     }
 
     [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
+    [CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)]
     public async Task ConsumerAsync(TrackPushEvent @event)
     {
         var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.SubSystemId, @event.CustomId);
         if (pushConfigs == null || pushConfigs.Count == 0) return;
         _pushConfigs = pushConfigs;
 
-        var errorType = (int)TrackProblemType.Error;
         var orders = await _orderRepository.GetListAsync(x =>
-            @event.OrderIds.Contains(x.Id) && x.QuestionType != errorType, x => new OrderEntity
+            @event.OrderIds.Contains(x.Id), x => new OrderEntity
         {
             Id = x.Id, OrderNo = x.OrderNo, LatestPushTrackCode = x.LatestPushTrackCode,
             OrderType = x.OrderType
@@ -80,6 +80,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
         {
             foreach (var order in _orders)
             {
+                if (order.QuestionType == (int)TrackProblemType.Error) continue; // 系统不推异常单，只能人工操作
                 // 推送新增，需要根据上一次的推送 推断 本次推送 是否可行
                 config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode);
                 if (config == null) continue;
@@ -114,6 +115,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
 
             if (tracks.Count == 0) return; // 一个能推的轨迹都不存在
 
+            // tips：人工推送只要能推就推了，不会校验已推和顺序问题，因为可能是更新
             foreach (var track in tracks)
             {
                 var order = _orders.First(x => x.Id == track.OrderId);
@@ -144,12 +146,18 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
             }
         }
 
-        string asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString());
-
         if (items.Count == 0) return;
-        await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.PushTrackInterfaceCode,
+
+        string asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString());
+        var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
+            InterfaceRelayConst.PushTrackInterfaceCode,
             _pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe,
             attach: asciiString);
+
+        if (r.Code != InterfaceRelayConst.SuccessCode)
+        {
+            throw new BusinessException(message: $"requestId: {r.RequestId}; {r.Msg}");
+        }
     }
 
     private async Task DoUpdateAsync(TrackPushEvent @event)
@@ -179,9 +187,17 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
         }
 
         if (items.Count == 0) return;
-        await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(InterfaceRelayConst.TrackUpdateInterfaceCode,
+
+        string asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString());
+        var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
+            InterfaceRelayConst.TrackUpdateInterfaceCode,
             _pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe,
-            attach: OptionType.修改.ToString());
+            attach: asciiString);
+
+        if (r.Code != InterfaceRelayConst.SuccessCode)
+        {
+            throw new BusinessException(message: $"requestId: {r.RequestId}; {r.Msg}");
+        }
     }
 
     private TrackUpdateInput BuiltUpdateItem(TrackPushContentConfigDto config, TrackPushRecordIndex log)

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

@@ -14,7 +14,7 @@ using SJZY.Track.Domain.Consts;
 using SJZY.Track.Domain.Enums;
 using Volo.Abp.DependencyInjection;
 
-namespace SJZY.Track.Application.EventHandlers;
+namespace SJZY.Track.Application.CAP.EventHandlers;
 
 public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
 {

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

@@ -8,7 +8,7 @@ using SJZY.Track.Domain.Consts;
 using SJZY.Track.Domain.Queries;
 using Volo.Abp.DependencyInjection;
 
-namespace SJZY.Track.Application.EventHandlers;
+namespace SJZY.Track.Application.CAP.EventHandlers;
 
 public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
 {
@@ -104,9 +104,10 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
     private async Task HandleTrackAsync(TrackUpsertEvent @event, List<int> canNextOrderIds)
     {
         var orderIds = @event.OrderIdentifiers.Select(x => x.OrderId).ToArray();
-        var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode);
+        var tracks = await _trackDetailsRepository.GetListAsync(orderIds, @event.TrackCode, true);
 
-        TrackDetailsIndex? firstTrack = null;
+        var newOrderIds = orderIds.Except(tracks.Select(x => x.OrderId)).ToArray();
+        var trackContents = await _provideQueries.MatchTrackContentAsync(newOrderIds, @event.TrackCode);
 
         var insertTracks = new List<TrackDetailsIndex>();
 
@@ -136,9 +137,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
                     CreateTime = DateTime.Now
                 };
 
-                //todo：可以优化?
                 // 根据轨迹code在订单模板内容中寻找匹配的
-                var trackContent = await _provideQueries.MatchTrackContentAsync(order.OrderId, @event.TrackCode);
+                var trackContent = trackContents.FirstOrDefault(x => x.OrderId == order.OrderId);
                 if (trackContent == default)
                 {
                     // 订单所绑定的模板，没有一条内容可以匹配发生的轨迹
@@ -166,30 +166,32 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
             }
 
             canNextOrderIds.Add(order.OrderId);
-
-            firstTrack ??= track;
         }
 
+        if (canNextOrderIds.Count == 0) return;
+
+        var content = await _provideQueries.GetTrackContentAsync(@event.SubSystemId, @event.TrackCode);
+
         // 批量add to es
         await _trackDetailsRepository.AddManyAsync(insertTracks);
 
-        if (firstTrack == null) return;
-
         // 触发订单主表更新事件（异步）
         await _capPublisher.PublishAsync(GlobalConst.TrackChangeAfterUpdateOrderSubscribe,
             new TrackChangeAfterUpdateOrderEvent
             {
-                OrderIds = orderIds,
+                OrderIds = canNextOrderIds.ToArray(),
                 IsCompleted = @event.TrackRefreshData.IsCompleted,
-                TrackContent = firstTrack.TrackContent,
+                TrackContent = content.TrackContent,
-                TrackEnContent = firstTrack.TrackEnContent,
+                TrackEnContent = content.TrackEnContent,
                 TrackPlace = @event.TrackPlace,
                 TrackTime = @event.TrackTime,
                 TrackTimeZone = @event.TrackTimeZone
             });
 
         // 触发轨迹推送事件（异步）
-        foreach (var group in @event.OrderIdentifiers.GroupBy(x => x.CustomId))
+        foreach (var group in @event.OrderIdentifiers
+                     .Where(x => x.OrderId.IsIn(canNextOrderIds))
+                     .GroupBy(x => x.CustomId))
         {
             var trackPushEvent = new TrackPushEvent
             {
@@ -200,8 +202,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
                 TrackCode = @event.TrackCode,
                 SystemPushInfo = new SystemPushStruct
                 {
-                    TrackStatus = firstTrack.TrackStatus,
+                    TrackStatus = content.TrackStatus,
-                    TrackContent = firstTrack.TrackContent,
+                    TrackContent = content.TrackContent,
                     TrackPlace = @event.TrackPlace,
                     TrackTime = @event.TrackTime,
                     TrackTimeZone = @event.TrackTimeZone

src/SJZY.Track.Application/CAP/LoggerExtensions.cs

@@ -0,0 +1,89 @@
+using Microsoft.Extensions.Logging;
+
+namespace SJZY.Track.Application.CAP;
+
+internal static class LoggerExtensions
+{
+    public static void ConsumerExecutedAfterThreshold(this ILogger logger, string messageId, int retries)
+    {
+        logger.LogWarning(
+            $"The Subscriber of the message({messageId}) still fails after {retries}th executions and we will stop retrying.");
+    }
+
+    public static void SenderAfterThreshold(this ILogger logger, string messageId, int retries)
+    {
+        logger.LogWarning(
+            $"The Publisher of the message({messageId}) still fails after {retries}th sends and we will stop retrying.");
+    }
+
+    public static void ExecutedThresholdCallbackFailed(this ILogger logger, Exception ex)
+    {
+        logger.LogWarning(ex, "FailedThresholdCallback action raised an exception:" + ex.Message);
+    }
+
+    public static void ConsumerDuplicates(this ILogger logger, string subscriber, string group)
+    {
+        logger.LogWarning(
+            $"We detected that you have duplicate subscribers ({subscriber}) in same group ({group}), this will cause diversity behavior.");
+    }
+
+    public static void ConsumerExecutionRetrying(this ILogger logger, string messageId, int retries)
+    {
+        logger.LogWarning($"The {retries}th retrying consume a message failed. message id: {messageId}");
+    }
+
+    public static void SenderRetrying(this ILogger logger, string messageId, int retries)
+    {
+        logger.LogWarning($"The {retries}th retrying send a message failed. message id: {messageId} ");
+    }
+
+    public static void MessageReceived(this ILogger logger, string messageId, string name)
+    {
+        logger.LogDebug($"Received message. id:{messageId}, name: {name}");
+    }
+
+    public static void MessagePublishException(this ILogger logger, string? messageId, string reason, Exception? ex)
+    {
+        logger.LogError(ex,
+            $"An exception occurred while publishing a message, reason:{reason}. message id:{messageId}");
+    }
+
+    public static void ConsumerExecuting(this ILogger logger, string className, string methodName, string group)
+    {
+        logger.LogInformation($"Executing subscriber method '{className}.{methodName}' on group '{group}'");
+    }
+
+    public static void ConsumerExecuted(this ILogger logger, string className, string methodName, string group,
+        double milliseconds, string? instance)
+    {
+        logger.LogInformation(
+            $"Executed subscriber method '{className}.{methodName}' on group '{group}' with instance '{instance}' in {milliseconds}ms");
+    }
+
+    public static void ConsumerExecuteFailed(this ILogger logger, string topic, string id, string? instance,
+        Exception? ex)
+    {
+        logger.LogError(ex,
+            $"An exception occurred while executing the subscription method. Topic:{topic}, Id:{id}, Instance: {instance}");
+    }
+
+    public static void ServerStarting(this ILogger logger)
+    {
+        logger.LogInformation("Starting the processing server.");
+    }
+
+    public static void ProcessorsStartedError(this ILogger logger, Exception ex)
+    {
+        logger.LogError(ex, "Starting the processors throw an exception.");
+    }
+
+    public static void ServerShuttingDown(this ILogger logger)
+    {
+        logger.LogInformation("Shutting down the processing server...");
+    }
+
+    public static void ExpectedOperationCanceledException(this ILogger logger, Exception ex)
+    {
+        logger.LogWarning(ex, $"Expected an OperationCanceledException, but found '{ex.Message}'.");
+    }
+}

src/SJZY.Track.Application/CAP/MySubscribeExecutor.cs

@@ -0,0 +1,268 @@
+using System.Diagnostics;
+using System.Reflection;
+using DotNetCore.CAP;
+using DotNetCore.CAP.Diagnostics;
+using DotNetCore.CAP.Internal;
+using DotNetCore.CAP.Messages;
+using DotNetCore.CAP.Persistence;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+namespace SJZY.Track.Application.CAP;
+
+public class MySubscribeExecutor : ISubscribeExecutor
+{
+    // diagnostics listener
+    // ReSharper disable once InconsistentNaming
+    private static readonly DiagnosticListener s_diagnosticListener =
+        new(CapDiagnosticListenerNames.DiagnosticListenerName);
+
+    private readonly IDataStorage _dataStorage;
+    private readonly string? _hostName;
+    private readonly ILogger _logger;
+    private readonly CapOptions _options;
+    private readonly IServiceProvider _provider;
+
+    public MySubscribeExecutor(
+        ILogger<MySubscribeExecutor> logger,
+        IOptions<CapOptions> options,
+        IServiceProvider provider)
+    {
+        _provider = provider;
+        _logger = logger;
+        _options = options.Value;
+
+        _dataStorage = _provider.GetRequiredService<IDataStorage>();
+        Invoker = _provider.GetRequiredService<ISubscribeInvoker>();
+        _hostName = Helper.GetInstanceHostname();
+    }
+
+    private ISubscribeInvoker Invoker { get; }
+
+    public async Task<OperateResult> ExecuteAsync(MediumMessage message, ConsumerExecutorDescriptor? descriptor = null,
+        CancellationToken cancellationToken = default)
+    {
+        if (descriptor == null)
+        {
+            var selector = _provider.GetRequiredService<MethodMatcherCache>();
+            if (!selector.TryGetTopicExecutor(message.Origin.GetName(), message.Origin.GetGroup()!, out descriptor))
+            {
+                var error =
+                    $"Message (Name:{message.Origin.GetName()},Group:{message.Origin.GetGroup()}) can not be found subscriber." +
+                    $"{Environment.NewLine} see: https://github.com/dotnetcore/CAP/issues/63";
+                _logger.LogError(error);
+
+                TracingError(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), message.Origin, null,
+                    new Exception(error));
+
+                var ex = new SubscriberNotFoundException(error);
+                await SetFailedState(message, ex);
+                return OperateResult.Failed(ex);
+            }
+        }
+
+        bool retry;
+        OperateResult result;
+
+        //record instance id
+        message.Origin.Headers[Headers.ExecutionInstanceId] = _hostName;
+
+        do
+        {
+            var (shouldRetry, operateResult) = await ExecuteWithoutRetryAsync(message, descriptor, cancellationToken)
+                .ConfigureAwait(false);
+            result = operateResult;
+            if (result.Equals(OperateResult.Success)) return result;
+            retry = shouldRetry;
+        } while (retry);
+
+        return result;
+    }
+
+    private async Task<(bool, OperateResult)> ExecuteWithoutRetryAsync(MediumMessage message,
+        ConsumerExecutorDescriptor descriptor, CancellationToken cancellationToken)
+    {
+        if (message == null) throw new ArgumentNullException(nameof(message));
+
+        cancellationToken.ThrowIfCancellationRequested();
+
+        try
+        {
+            _logger.ConsumerExecuting(descriptor.ImplTypeInfo.Name, descriptor.MethodInfo.Name,
+                descriptor.Attribute.Group);
+
+            var sp = Stopwatch.StartNew();
+
+            await InvokeConsumerMethodAsync(message, descriptor, cancellationToken).ConfigureAwait(false);
+
+            sp.Stop();
+
+            await SetSuccessfulState(message).ConfigureAwait(false);
+
+            CapEventCounterSource.Log.WriteInvokeTimeMetrics(sp.Elapsed.TotalMilliseconds);
+            _logger.ConsumerExecuted(descriptor.ImplTypeInfo.Name, descriptor.MethodInfo.Name,
+                descriptor.Attribute.Group, sp.Elapsed.TotalMilliseconds, message.Origin.GetExecutionInstanceId());
+
+            return (false, OperateResult.Success);
+        }
+        catch (Exception ex)
+        {
+            _logger.ConsumerExecuteFailed(message.Origin.GetName(), message.DbId,
+                message.Origin.GetExecutionInstanceId(), ex);
+
+            return (await SetFailedState(message, ex).ConfigureAwait(false), OperateResult.Failed(ex));
+        }
+    }
+
+    private Task SetSuccessfulState(MediumMessage message)
+    {
+        message.ExpiresAt = DateTime.Now.AddSeconds(_options.SucceedMessageExpiredAfter);
+
+        return _dataStorage.ChangeReceiveStateAsync(message, StatusName.Succeeded);
+    }
+
+    private async Task<bool> SetFailedState(MediumMessage message, Exception ex)
+    {
+        if (ex is SubscriberNotFoundException)
+            message.Retries = _options.FailedRetryCount; // not retry if SubscriberNotFoundException
+
+        var needRetry = UpdateMessageForRetry(message);
+
+        message.Origin.AddOrUpdateException(ex);
+        message.ExpiresAt = message.Added.AddSeconds(_options.FailedMessageExpiredAfter);
+
+        await _dataStorage.ChangeReceiveStateAsync(message, StatusName.Failed).ConfigureAwait(false);
+
+        return needRetry;
+    }
+
+    private bool UpdateMessageForRetry(MediumMessage message)
+    {
+        var retries = ++message.Retries;
+
+        if (retries == _options.FailedRetryCount)
+        {
+            try
+            {
+                _options.FailedThresholdCallback?.Invoke(new FailedInfo
+                {
+                    ServiceProvider = _provider,
+                    MessageType = MessageType.Subscribe,
+                    Message = message.Origin
+                });
+
+                _logger.ConsumerExecutedAfterThreshold(message.DbId, _options.FailedRetryCount);
+            }
+            catch (Exception ex)
+            {
+                _logger.ExecutedThresholdCallbackFailed(ex);
+            }
+        }
+
+        return false;
+    }
+
+    private async Task InvokeConsumerMethodAsync(MediumMessage message, ConsumerExecutorDescriptor descriptor,
+        CancellationToken cancellationToken)
+    {
+        var consumerContext = new ConsumerContext(descriptor, message);
+        var tracingTimestamp = TracingBefore(message.Origin, descriptor.MethodInfo);
+        try
+        {
+            var ret = await Invoker.InvokeAsync(consumerContext, cancellationToken).ConfigureAwait(false);
+
+            TracingAfter(tracingTimestamp, message.Origin, descriptor.MethodInfo);
+
+            if (!string.IsNullOrEmpty(ret.CallbackName))
+            {
+                var header = new Dictionary<string, string?>
+                {
+                    [Headers.CorrelationId] = message.Origin.GetId(),
+                    [Headers.CorrelationSequence] = (message.Origin.GetCorrelationSequence() + 1).ToString()
+                };
+
+                if (message.Origin.Headers.TryGetValue(Headers.TraceParent, out var traceparent))
+                    header[Headers.TraceParent] = traceparent;
+
+                await _provider.GetRequiredService<ICapPublisher>()
+                    .PublishAsync(ret.CallbackName, ret.Result, header, cancellationToken).ConfigureAwait(false);
+            }
+        }
+        catch (OperationCanceledException)
+        {
+            //ignore
+        }
+        catch (Exception ex)
+        {
+            var e = new SubscriberExecutionFailedException(ex.Message, ex);
+
+            TracingError(tracingTimestamp, message.Origin, descriptor.MethodInfo, e);
+
+            e.ReThrow();
+        }
+    }
+
+    #region tracing
+
+    private long? TracingBefore(Message message, MethodInfo method)
+    {
+        if (s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.BeforeSubscriberInvoke))
+        {
+            var eventData = new CapEventDataSubExecute
+            {
+                OperationTimestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
+                Operation = message.GetName(),
+                Message = message,
+                MethodInfo = method
+            };
+
+            s_diagnosticListener.Write(CapDiagnosticListenerNames.BeforeSubscriberInvoke, eventData);
+
+            return eventData.OperationTimestamp;
+        }
+
+        return null;
+    }
+
+    private void TracingAfter(long? tracingTimestamp, Message message, MethodInfo method)
+    {
+        CapEventCounterSource.Log.WriteInvokeMetrics();
+        if (tracingTimestamp != null &&
+            s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.AfterSubscriberInvoke))
+        {
+            var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+            var eventData = new CapEventDataSubExecute
+            {
+                OperationTimestamp = now,
+                Operation = message.GetName(),
+                Message = message,
+                MethodInfo = method,
+                ElapsedTimeMs = now - tracingTimestamp.Value
+            };
+
+            s_diagnosticListener.Write(CapDiagnosticListenerNames.AfterSubscriberInvoke, eventData);
+        }
+    }
+
+    private void TracingError(long? tracingTimestamp, Message message, MethodInfo? method, Exception ex)
+    {
+        if (tracingTimestamp != null &&
+            s_diagnosticListener.IsEnabled(CapDiagnosticListenerNames.ErrorSubscriberInvoke))
+        {
+            var now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+            var eventData = new CapEventDataSubExecute
+            {
+                OperationTimestamp = now,
+                Operation = message.GetName(),
+                Message = message,
+                MethodInfo = method,
+                ElapsedTimeMs = now - tracingTimestamp.Value,
+                Exception = ex
+            };
+
+            s_diagnosticListener.Write(CapDiagnosticListenerNames.ErrorSubscriberInvoke, eventData);
+        }
+    }
+
+    #endregion
+}

src/SJZY.Track.Application/CAP/SubscriberExecutionFailedException.cs

@@ -0,0 +1,15 @@
+namespace SJZY.Track.Application.CAP;
+
+internal class SubscriberExecutionFailedException : Exception
+{
+    private readonly Exception _originException;
+
+    public SubscriberExecutionFailedException(string message, Exception ex) : base(message, ex)
+    {
+        _originException = ex;
+    }
+
+    public override string? StackTrace => _originException.StackTrace;
+
+    public override string? Source => _originException.Source;
+}

src/SJZY.Track.Application/EventHandlers/CAPFilter.cs

@@ -1,121 +0,0 @@
-using System.Globalization;
-using DotNetCore.CAP.Filter;
-using SJZY.Expand.ABP.Core;
-using SJZY.Expand.ABP.Core.Logger;
-using SJZY.Expand.ABP.Core.PublicMethords;
-using SJZY.Track.Domain.Consts;
-
-namespace SJZY.Track.Application.EventHandlers;
-
-public class CapFilter : SubscribeFilter
-{
-    const string ErrorCode = "Error599";
-
-    /// <summary>
-    /// 执行前
-    /// </summary>
-    /// <param name="context"></param>
-    /// <returns></returns>
-    public override Task OnSubscribeExecutingAsync(ExecutingContext context)
-    {
-        context.DeliverMessage.Headers["sjzy-receive-time"] = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss fff");
-
-        // 轨迹增改要做幂等
-        if (!context.DeliverMessage.Headers["cap-msg-name"].IsIn(GlobalConst.TrackUpsertSubscribe))
-            return Task.CompletedTask;
-
-        // 本地调试不做幂等
-        if (System.Diagnostics.Debugger.IsAttached)
-        {
-            return Task.CompletedTask;
-        }
-
-        var argumentJsonStr = context.Arguments.ToJson();
-        context.DeliverMessage.Headers["sjzy-arguments"] = argumentJsonStr;
-
-        if (!string.IsNullOrEmpty(argumentJsonStr))
-        {
-            var hash = argumentJsonStr.GetHash();
-            using (RedisHelper.Client.Lock("Filter:" + hash, 10))
-            {
-                var isSkip = !string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExcuteFilter:{hash}"));
-                if (isSkip)
-                    throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}");
-
-                RedisHelper.Client.Set($"CapExcuteFilter:{hash}", 60 * 10,
-                    TimeSpan.FromMinutes(5).Seconds);
-            }
-        }
-
-        return Task.CompletedTask;
-    }
-
-    /// <summary>
-    /// 执行后
-    /// </summary>
-    /// <param name="context"></param>
-    /// <returns></returns>
-    public override Task OnSubscribeExecutedAsync(ExecutedContext context)
-    {
-        // context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime);
-        // context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName);
-        // context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments);
-        // context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
-        //
-        // BuildLog(receiveTime!, receiveName!, arguments!, false, messageId!);
-
-        return Task.CompletedTask;
-    }
-
-
-    /// <summary>
-    /// 消费异常事件
-    /// </summary>
-    /// <param name="context"></param>
-    /// <returns></returns>
-    public override Task OnSubscribeExceptionAsync(ExceptionContext context)
-    {
-        context.DeliverMessage.Headers.TryGetValue("cap-msg-id", out var messageId);
-
-        if (context.Exception is BusinessException ex && ex.Code == ErrorCode)
-        {
-            context.ExceptionHandled = true; // 幂等不做重试
-            LogHelper.Warn(ex.ToString());
-        }
-        else
-            LogHelper.Error($"cap-msg-id：{messageId}\n{context.Exception}");
-
-
-        // context.DeliverMessage.Headers.TryGetValue("sjzy-receive-time", out var receiveTime);
-        // context.DeliverMessage.Headers.TryGetValue("cap-msg-name", out var receiveName);
-        // context.DeliverMessage.Headers.TryGetValue("sjzy-arguments", out var arguments);
-        //
-        // BuildLog(receiveTime!, receiveName!, arguments, true, messageId!);
-
-        return Task.CompletedTask;
-    }
-
-
-    private void BuildLog(string receiveTime, string receiveName, string? arguments, bool isFromExecptionFliter,
-        string messageId)
-    {
-        var parsedDate = DateTime.ParseExact(receiveTime, "yyyy-MM-dd HH:mm:ss fff", CultureInfo.InvariantCulture);
-        var useTime = DateTime.Now - parsedDate;
-
-        if (useTime.Minutes > 10)
-        {
-            var log = $@"
-|-消息编号：{messageId}
-|-消费者名：{receiveName}
-|-跳过执行：{(isFromExecptionFliter ? "是" : "否")}
-|-执行时间：{receiveTime}
-|-消费耗时：{useTime.Seconds}秒
-|-请求参数：{arguments}";
-            LogHelper.Error(log);
-        }
-        else
-        {
-            //LogHelper.Info(log);
-        }
-    }
-}

src/SJZY.Track.Application/SJZY.Track.Application.csproj

@@ -12,15 +12,17 @@
     </ItemGroup>
 
     <ItemGroup>
-        <PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.2.0"/>
+        <PackageReference Include="SJZY.InterfaceRelay.Application.Contracts" Version="7.2.1"/>
         <PackageReference Include="Volo.Abp.Http.Client" Version="7.0.2" />
         <PackageReference Include="Volo.Abp.FluentValidation" Version="7.0.2" />
         <PackageReference Include="SJZY.UserCenter.HttpApi" Version="3.0.9-alpha"/>
         <PackageReference Include="SJZY.Expand.CAP" Version="7.8.0" />
-        <PackageReference Include="DotNetCore.CAP" Version="8.0.1" />
+        <PackageReference Include="DotNetCore.CAP" Version="8.1.2"/>
-        <PackageReference Include="DotNetCore.CAP.Dashboard" Version="8.0.1" />
+        <PackageReference Include="DotNetCore.CAP.Dashboard" Version="8.1.2"/>
-        <PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="8.0.1" />
+        <PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="8.1.2"/>
-        <PackageReference Include="DotNetCore.CAP.MySql" Version="8.0.1" />
+        <PackageReference Include="DotNetCore.CAP.MySql" Version="8.1.2"/>
+        <PackageReference Include="DotNetCore.CAP.InMemoryStorage" Version="8.1.2"/>
+        <PackageReference Include="Savorboard.CAP.InMemoryMessageQueue" Version="7.0.0"/>
     </ItemGroup>
 
     <ItemGroup>

src/SJZY.Track.Application/Services/ProvideServices/TrackPullService.cs

@@ -78,46 +78,50 @@ public class TrackPullService : ApplicationService
     /// <returns></returns>
     [AllowAnonymous]
     [HttpPost("Track/RefreshForTask")]
-    public async Task<bool> RefreshForTaskAsync()
+    public bool RefreshForTaskAsync()
     {
-        var configs = await _trackPullCacheService.GetAllPullConfigsAsync();
+        Task.Run(async () =>
-        foreach (var config in configs)
         {
-            var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderId);
+            var configs = await _trackPullCacheService.GetAllPullConfigsAsync();
-            if (await RedisHelper.Client.ExistsAsync(redisKey))
-            {
-                continue;
-            }
 
-            var page = 1;
+            foreach (var config in configs)
-            var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
-            while (orders.Any())
             {
-                var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
+                var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderId);
-                var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName;
+                if (await RedisHelper.Client.ExistsAsync(redisKey))
-
-                foreach (var group in orders.GroupBy(x =>
-                             new { x.SubSystemId, x.OrderType }))
                 {
-                    var data = group.Select(x => new
+                    continue;
-                    {
-                        x.OrderNo
-                    }).ToList();
-
-                    var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId,
-                        serviceProviderName!,
-                        group.Key.SubSystemId, group.Key.OrderType));
-                    // 批量获取轨迹
-                    await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
-                        InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data,
-                        GlobalConst.TrackRefreshSubscribe, attach);
                 }
 
-                orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page);
+                var page = 1;
-            }
+                var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
+                while (orders.Any())
+                {
+                    var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
+                    var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName;
 
-            await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count);
+                    foreach (var group in orders.GroupBy(x =>
-        }
+                                 new { x.SubSystemId, x.OrderType }))
+                    {
+                        var data = group.Select(x => new
+                        {
+                            x.OrderNo
+                        }).ToList();
+
+                        var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId,
+                            serviceProviderName!,
+                            group.Key.SubSystemId, group.Key.OrderType));
+                        // 批量获取轨迹
+                        await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
+                            InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data,
+                            GlobalConst.TrackRefreshSubscribe, attach);
+                    }
+
+                    orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page);
+                }
+
+                await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count);
+            }
+        });
 
         return true;
     }

src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs

@@ -6,6 +6,7 @@ using SJZY.Track.Domain.Aggregates.OrderAggregate;
 using SJZY.Track.Domain.Aggregates.TrackPushAggregate;
 using SJZY.Track.Domain.Consts;
 using SJZY.UserCenter.HttpApi.Services;
+using Volo.Abp.Http.Client.DynamicProxying;
 
 namespace SJZY.Track.Application.Services.ProvideServices;
 
@@ -19,14 +20,17 @@ public class TrackPushService : ApplicationService
     private readonly IOrderRepository _orderRepository;
     private readonly ICapPublisher _capPublisher;
     private readonly ITrackPushRecordsRepository _trackPushRecordsRepository;
+    private readonly IHttpClientProxy<IRequestCurrentUserService> _requestCurrentUserService;
 
     public TrackPushService(IOrderRepository orderRepository
         , ICapPublisher capPublisher
-        , ITrackPushRecordsRepository trackPushRecordsRepository)
+        , ITrackPushRecordsRepository trackPushRecordsRepository,
+        IHttpClientProxy<IRequestCurrentUserService> requestCurrentUserService)
     {
         _orderRepository = orderRepository;
         _capPublisher = capPublisher;
         _trackPushRecordsRepository = trackPushRecordsRepository;
+        _requestCurrentUserService = requestCurrentUserService;
     }
 
     /// <summary>
@@ -39,25 +43,42 @@ public class TrackPushService : ApplicationService
     {
         input.OrderNos = input.OrderNos.Where(x => !x.IsNullOrWhiteSpace()).Distinct().ToList(); // 单号去重、去空
 
-        var orderIds = await _orderRepository.GetArrayAsync(x =>
+        var orders = await _orderRepository.GetArrayAsync(x =>
-            x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => x.Id);
+            x.OrderType == input.OrderType && input.OrderNos.Contains(x.OrderNo), x => new
+        {
+            x.Id, x.SubSystemId, x.CustomId
+        });
 
-        if (!orderIds.Any())
+        if (!orders.Any())
         {
             throw new BusinessException(message: "一个订单都不存在");
         }
 
-        var trackPushEvent = new TrackPushEvent
+        var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
+        if (userInfo.Code != "200" || userInfo.Data is null)
         {
-            OptionType = OptionType.新增,
+            //TODO: log || ex
-            OrderIds = orderIds,
+            throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}");
-            TrackCode = input.TrackCode,
+        }
-            OperationMan = CurrentUserRequest.CurrentUser.Name,
-        };
 
-        await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
+        var name = userInfo.Data.UserName;
 
-        return $"即将推送轨迹代码{input.TrackCode}给{orderIds.Length}个订单";
+        foreach (var group in orders.GroupBy(x => new { x.SubSystemId, x.CustomId }))
+        {
+            var trackPushEvent = new TrackPushEvent
+            {
+                OptionType = OptionType.新增,
+                SubSystemId = group.Key.SubSystemId,
+                CustomId = group.Key.CustomId,
+                OrderIds = group.Select(x => x.Id).ToArray(),
+                TrackCode = input.TrackCode,
+                OperationMan = name,
+            };
+
+            await _capPublisher.PublishAsync(GlobalConst.TrackManualPushSubscribe, trackPushEvent);
+        }
+
+        return $"即将推送轨迹代码{input.TrackCode}给{orders.Length}个订单";
     }
 
 

src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs

@@ -7,6 +7,7 @@ using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
 using SJZY.Track.Domain.Consts;
 using SJZY.Track.Domain.Queries;
 using SJZY.UserCenter.HttpApi.Services;
+using Volo.Abp.Http.Client.DynamicProxying;
 
 namespace SJZY.Track.Application.Services.ProvideServices;
 
@@ -20,14 +21,17 @@ public partial class TrackService : ApplicationService, ITrackService
     private readonly IOrderRepository _orderRepository;
     private readonly IProvideQueries _provideQueries;
     private readonly ICapPublisher _capPublisher;
+    private readonly IHttpClientProxy<IRequestCurrentUserService> _requestCurrentUserService;
 
     public TrackService(IOrderRepository orderRepository, IProvideQueries provideQueries, ICapPublisher capPublisher,
-        ITrackDetailsRepository trackDetailsRepository)
+        ITrackDetailsRepository trackDetailsRepository,
+        IHttpClientProxy<IRequestCurrentUserService> requestCurrentUserService)
     {
         _orderRepository = orderRepository;
         _provideQueries = provideQueries;
         _provideQueries = provideQueries;
         _capPublisher = capPublisher;
+        _requestCurrentUserService = requestCurrentUserService;
         _trackDetailsRepository = trackDetailsRepository;
     }
 
@@ -50,6 +54,15 @@ public partial class TrackService : ApplicationService, ITrackService
             throw new BusinessException(message: $"订单：{input.OrderNo}不存在");
         }
 
+        var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
+        if (userInfo.Code != "200" || userInfo.Data is null)
+        {
+            //TODO: log || ex
+            throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}");
+        }
+
+        var name = userInfo.Data.UserName;
+
         // 新增/更新轨迹
         var @event = ObjectMapper.Map<UpsertTrackInput, TrackUpsertEvent>(input);
         @event.SubSystemId = order.SubSystemId;
@@ -63,7 +76,7 @@ public partial class TrackService : ApplicationService, ITrackService
                 CustomId = order.CustomId
             }
         };
-        @event.OperationMan = CurrentUserRequest.CurrentUser.Name;
+        @event.OperationMan = name;
         await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event);
         return new UnifyResultDto(true, true);
     }
@@ -89,6 +102,15 @@ public partial class TrackService : ApplicationService, ITrackService
             x.Id, x.OrderNo, x.SubSystemId, x.CustomId
         });
 
+        var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
+        if (userInfo.Code != "200" || userInfo.Data is null)
+        {
+            //TODO: log || ex
+            throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}");
+        }
+
+        var name = userInfo.Data.UserName;
+
         foreach (var group in orders.GroupBy(x => x.SubSystemId))
         {
             // 新增/更新轨迹
@@ -107,7 +129,7 @@ public partial class TrackService : ApplicationService, ITrackService
                 TrackPlace = input.TrackPlace,
                 TrackTime = input.TrackTime!.Value,
                 TrackTimeZone = input.TrackTimeZone!.Value,
-                OperationMan = CurrentUserRequest.CurrentUser.Name
+                OperationMan = name
             };
             await _capPublisher.PublishAsync(GlobalConst.TrackUpsertSubscribe, @event);
         }
@@ -143,13 +165,22 @@ public partial class TrackService : ApplicationService, ITrackService
             throw new BusinessException(message: "没有一个轨迹代码是可删的");
         }
 
+        var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
+        if (userInfo.Code != "200" || userInfo.Data is null)
+        {
+            //TODO: log || ex
+            throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}");
+        }
+
+        var name = userInfo.Data.UserName;
+
         // 删除轨迹
         var @event = new TrackDeleteEvent
         {
             OrderIds = new[] { order.Id },
             TrackCodes = codes.Select(x => x.TrackCode).ToList(),
             SubSystemId = order.SubSystemId,
-            OperationMan = CurrentUserRequest.CurrentUser.Name
+            OperationMan = name
         };
         await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, @event);
 
@@ -182,6 +213,15 @@ public partial class TrackService : ApplicationService, ITrackService
             throw new BusinessException(message: "没有一个轨迹代码是可删的");
         }
 
+        var userInfo = await _requestCurrentUserService.Service.GetCurrentUserBasicInfoAsync();
+        if (userInfo.Code != "200" || userInfo.Data is null)
+        {
+            //TODO: log || ex
+            throw new BusinessException(message: $"调用用户中心，获取用户信息失败:{userInfo?.Msg}");
+        }
+
+        var name = userInfo.Data.UserName;
+
         // 删除轨迹
         foreach (var group in codes.GroupBy(x => x.OrderId))
         {
@@ -190,7 +230,7 @@ public partial class TrackService : ApplicationService, ITrackService
                 OrderIds = group.Select(x => x.OrderId).ToArray(),
                 TrackCodes = input.TrackCodes,
                 SubSystemId = orders.First(x => x.Id == group.Key).SubSystemId,
-                OperationMan = CurrentUserRequest.CurrentUser.Name
+                OperationMan = name
             };
             await _capPublisher.PublishAsync(GlobalConst.TrackDeleteSubscribe, @event);
         }

src/SJZY.Track.Application/SjzyTrackApplicationModule.cs

@@ -1,12 +1,13 @@
-using System.Text.Json;
+using DotNetCore.CAP;
-using DotNetCore.CAP;
+using DotNetCore.CAP.Internal;
 using Microsoft.Extensions.Configuration;
+using Savorboard.CAP.InMemoryMessageQueue;
 using SJZY.Expand.ABP.Core.Localization;
 using SJZY.Expand.ABP.Core.Logger;
 using SJZY.Expand.ABP.Core.PublicMethords;
 using SJZY.InterfaceRelay.Application.Contracts;
+using SJZY.Track.Application.CAP;
 using SJZY.Track.Application.Contract;
-using SJZY.Track.Application.EventHandlers;
 using SJZY.Track.Application.Localization;
 using SJZY.UserCenter.Contracts;
 using SJZY.UserCenter.SharedModel;
@@ -74,9 +75,14 @@ namespace SJZY.Track.Application
             );
 
             //CAP
+            context.Services.AddSingleton<ISubscribeExecutor, MySubscribeExecutor>();
             context.Services.AddCapTableNameService();
             context.Services.AddCap(options =>
             {
+#if DEBUG
+                options.UseInMemoryStorage();
+                options.UseInMemoryMessageQueue();
+#else
                 options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!);
                 options.UseRabbitMQ(configure =>
                 {
@@ -87,13 +93,13 @@ namespace SJZY.Track.Application
                     configure.BasicQosOptions =
                         new RabbitMQOptions.BasicQos(configuration.GetValue<ushort>("CAP:PrefetchCount"));
                 });
-                // options.TopicNamePrefix = "v1";
+#endif
                 options.FailedRetryInterval = configuration.GetValue<int>("CAP:FailedRetryInterval");
                 options.FailedRetryCount = configuration.GetValue<int>("CAP:FailedRetryCount");
-                options.ConsumerThreadCount = configuration.GetValue<int>("CAP:ConsumerThreadCount");
                 options.SucceedMessageExpiredAfter = configuration.GetValue<int>("CAP:SucceedMessageExpiredAfter");
-                options.UseDispatchingPerGroup = true;
+                options.FallbackWindowLookbackSeconds =
-                options.EnableConsumerPrefetch = true;
+                    configuration.GetValue<int>("CAP:FallbackWindowLookbackSeconds");
+                options.EnableSubscriberParallelExecute = true;
 
                 options.UseDashboard(dashboardOptions =>
                 {
@@ -108,8 +114,7 @@ namespace SJZY.Track.Application
                         LogHelper.Error($@"
     |-CAP失败回调事件：
     |-处理类型：{(faileCalback.MessageType == DotNetCore.CAP.Messages.MessageType.Subscribe ? "消费消息达到最大重试次数，需要开发人员介入" : "生产消息推送失败")}
-    |-头部信息：{faileCalback.Message.Headers.ToJson()}
+    |-头部信息：{faileCalback.Message.Headers.ToJson()}");
-   	|-消息内容：{(faileCalback.Message.Value is string ? faileCalback.Message.Value : ((JsonElement?)faileCalback.Message.Value)?.GetRawText())}");
                     }
                     catch (Exception ex)
                     {

src/SJZY.Track.Domain/Aggregates/TrackDetailsAggregate/ITrackDetailsRepository.cs

@@ -4,11 +4,10 @@ namespace SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
 
 public interface ITrackDetailsRepository
 {
-    Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode);
+    Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode, bool forUpsert = false);
 
     Task<List<TrackDetailsIndex>> GetLatestListAsync(int[] eventOrderIds, List<string> trackCodes);
 
-
     Task<bool> AnyAsync(int orderId, DateTime trackTime);
 
     Task<bool> AnyAsync(int orderId, string originTrackData);

src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs

@@ -32,6 +32,11 @@ public partial class GlobalConst
     /// </summary>
     public const string TrackPushSubscribe = "SJZY.Track.TrackPush";
 
+    /// <summary>
+    /// 轨迹人工手动推送
+    /// </summary>
+    public const string TrackManualPushSubscribe = "SJZY.Track.TrackManualPush";
+
     /// <summary>
     /// 轨迹推送完成（接口通回调）
     /// </summary>

src/SJZY.Track.Domain/Dtos/TrackContents/TrackContentVO.cs

@@ -0,0 +1,19 @@
+namespace SJZY.Track.Domain.Dtos.TrackContents;
+
+public class TrackContentVO
+{
+    /// <summary>
+    /// 轨迹状态
+    /// </summary>
+    public string TrackStatus { get; set; } = null!;
+
+    /// <summary>
+    /// 轨迹内容
+    /// </summary>
+    public string TrackContent { get; set; } = null!;
+
+    /// <summary>
+    /// 轨迹内容英文
+    /// </summary>
+    public string TrackEnContent { get; set; } = null!;
+}

src/SJZY.Track.Domain/Dtos/TrackDetails/MatchTrackContentDto.cs

@@ -3,9 +3,14 @@
 public class MatchTrackContentDto
 {
     /// <summary>
-    /// 轨迹内容id
+    /// 订单Id
     /// </summary>
-    public string TrackCode { get; set; }
+    public int OrderId { get; set; }
+
+    /// <summary>
+    /// 轨迹内容code
+    /// </summary>
+    public string TrackCode { get; set; } = null!;
 
     /// <summary>
     /// 轨迹内容

src/SJZY.Track.Domain/Queries/IProvideQueries.cs

@@ -1,4 +1,5 @@
 using SJZY.Track.Domain.Dtos.Order;
+using SJZY.Track.Domain.Dtos.TrackContents;
 using SJZY.Track.Domain.Dtos.TrackDetails;
 using SJZY.Track.Domain.Dtos.TrackTemplate;
 
@@ -27,10 +28,10 @@ public interface IProvideQueries
     /// <summary>
     /// 根据订单id找到订单模板，匹配模板命中的轨迹内容
     /// </summary>
-    /// <param name="orderId">订单id</param>
+    /// <param name="orderIds">订单ids</param>
     /// <param name="code">轨迹代码</param>
     /// <returns></returns>
-    Task<MatchTrackContentDto?> MatchTrackContentAsync(int orderId, string code);
+    Task<List<MatchTrackContentDto>> MatchTrackContentAsync(int[] orderIds, string code);
 
     /// <summary>
     /// 获取可删除的订单轨迹代码
@@ -65,4 +66,6 @@ public interface IProvideQueries
     /// <returns></returns>
     Task<List<OrderWithServiceProviderVO>> FindOrderWithServiceProviderAsync(int serviceProvider, int pageIndex,
         int pageSize = 1000);
+
+    Task<TrackContentVO> GetTrackContentAsync(int subSystemId, string trackCode);
 }

src/SJZY.Track.Repository/Queries/ProvideQueries.cs

@@ -3,6 +3,7 @@ using SJZY.Track.Domain.Aggregates.TrackContentAggregate;
 using SJZY.Track.Domain.Aggregates.TrackStatusAggregate;
 using SJZY.Track.Domain.Aggregates.TrackTemplateAggregate;
 using SJZY.Track.Domain.Dtos.Order;
+using SJZY.Track.Domain.Dtos.TrackContents;
 using SJZY.Track.Domain.Dtos.TrackDetails;
 using SJZY.Track.Domain.Dtos.TrackTemplate;
 using SJZY.Track.Domain.Queries;
@@ -50,22 +51,24 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
             .ToListAsync();
     }
 
-    public async Task<MatchTrackContentDto?> MatchTrackContentAsync(int orderId, string code)
+    public async Task<List<MatchTrackContentDto>> MatchTrackContentAsync(int[] orderIds, string code)
     {
+        if (orderIds.Length == 0) return new List<MatchTrackContentDto>();
         return await Context.Queryable<OrderTemplateEntity>()
             .LeftJoin<TrackTemplateContentEntity>((x, a) => x.TrackTemplateId == a.TrackTemplateId)
             .LeftJoin<TrackContentEntity>((x, a, b) => a.TrackContentId == b.Id)
             .LeftJoin<TrackStatusEntity>((x, a, b, c) => b.TrackStatusId == c.Id)
-            .Where((x, a, b) => x.OrderId == orderId && b.Code == code)
+            .Where((x, a, b) => orderIds.Contains(x.OrderId) && b.Code == code)
             .Select((x, a, b, c) => new MatchTrackContentDto
             {
+                OrderId = x.OrderId,
                 TrackCode = b.Code,
                 Content = b.Content,
                 EnContent = b.EnContent,
                 Status = c.Status,
                 IsCanDelete = a.IsCanDelete
             })
-            .FirstAsync();
+            .ToListAsync();
     }
 
     public async Task<List<CanDeleteTrackCodeVO>> GetCanDeleteTrackCodesAsync(List<int> orderIds,
@@ -133,4 +136,18 @@ public class ProvideQueries : ISugarRepository, IProvideQueries,
             .Take(pageSize)
             .ToListAsync();
     }
+
+    public async Task<TrackContentVO> GetTrackContentAsync(int subSystemId, string trackCode)
+    {
+        return await Context.Queryable<TrackContentEntity>()
+            .LeftJoin<TrackStatusEntity>((x, y) => x.TrackStatusId == y.Id)
+            .Where(x => x.SubSystemId == subSystemId && x.Code == trackCode)
+            .Select((x, y) => new TrackContentVO
+            {
+                TrackStatus = y.Status,
+                TrackContent = x.Content,
+                TrackEnContent = x.EnContent
+            })
+            .FirstAsync();
+    }
 }

src/SJZY.Track.Repository/Repositories/TrackDetailsRepository.cs

@@ -1,4 +1,3 @@
-using Elasticsearch.Net;
 using Nest;
 using SJZY.Expand.ElasticSearch.Services;
 using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
@@ -14,20 +13,37 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
     {
     }
 
-    public async Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode)
+    public async Task<List<TrackDetailsIndex>> GetListAsync(int[] orderIds, string trackCode,
+        bool forUpsert = false)
     {
-        var r = await Context.SearchAsync<TrackDetailsIndex>(s => s
+        var r = !forUpsert
-            .Source()
+            ? await Context.SearchAsync<TrackDetailsIndex>(s => s
-            .Size(1000)
+                .Source()
-            .Query(q => q
+                .Size(1000)
-                .Bool(b => b
+                .Query(q => q
-                    .Must(
+                    .Bool(b => b
-                        mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)),
+                        .Must(
-                        mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode))
+                            mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)),
+                            mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode))
+                        )
                     )
                 )
             )
-        );
+            : await Context.SearchAsync<TrackDetailsIndex>(s => s
+                .Source(s2 => s2.Includes(fd => fd
+                    .Field(f => f.IndexName)
+                    .Field(f => f.OrderId)
+                ))
+                .Size(1000)
+                .Query(q => q
+                    .Bool(b => b
+                        .Must(
+                            mu => mu.Terms(t => t.Field(f => f.OrderId).Terms(orderIds)),
+                            mu => mu.Term(t => t.Field(f => f.TrackCode.Suffix("keyword")).Value(trackCode))
+                        )
+                    )
+                )
+            );
 
         if (!r.IsValid)
         {
@@ -145,9 +161,16 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
 
     public async Task UpdateAsync(TrackDetailsIndex track)
     {
-        var r = await Context.UpdateAsync<TrackDetailsIndex>(track.Id, u => u
+        var r = await Context.UpdateAsync<TrackDetailsIndex, object>(track.Id, u => u
             .Index(track.IndexName)
-            .Doc(track)
+            .Doc(new
+            {
+                track.TrackPlace,
+                track.TrackTime,
+                track.TrackTimeZone,
+                track.UpdateUser,
+                UpdateTime = DateTime.Now,
+            })
         );
 
         if (!r.IsValid)
@@ -235,18 +258,28 @@ internal class TrackDetailsRepository : ElasticSearchContextAbstract<TrackDetail
 
     public async Task DeleteTrackAsync(int[] orderIds, List<string> trackCodes)
     {
-        var r = await Context.DeleteByQueryAsync<TrackDetailsIndex>(x => x
+        var search = await Context.SearchAsync<TrackDetailsIndex>(x => x
-            .Conflicts(Conflicts.Abort)
+            .Source(s => s.Includes(fd => fd.Field(f => f.IndexName)))
-            .Query(
+            .Size(orderIds.Length * trackCodes.Count * 2)
-                q =>
+            .Query(q =>
-                    q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
+                q.Terms(t => t.Field(f => f.OrderId).Terms(orderIds))
-                    && q.Term(t => t.IsCanDelete, true)
+                && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes))
-                    && q.Terms(t => t.Field(f => f.TrackCode.Suffix("keyword")).Terms(trackCodes)))
+            )
-            .WaitForCompletion()
         );
 
-        if (!r.IsValid)
+        if (!search.IsValid)
-            throw new Exception($"轨迹数据删除失败，原因：{r.DebugInformation}");
+        {
+            throw new Exception($"轨迹查询异常，原因：{search.DebugInformation}。");
+        }
+
+        var tracks = search.ToIdList();
+
+        foreach (var track in tracks)
+        {
+            await Context.DeleteAsync<TrackDetailsIndex>(track.Id, u => u
+                .Index(track.IndexName)
+            );
+        }
     }
 
     public async Task<(List<TrackDetailsIndex> list, long total)> GetListAsync(TrackDetailsQueryInput input)