opt：优化推送幂等，优化cap重试机制

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

@@ -1,10 +1,12 @@
-using DotNetCore.CAP.Filter;
+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.EventHandlers;
+namespace SJZY.Track.Application.CAP;
 
 public class CapFilter : SubscribeFilter
 {
@@ -17,21 +19,11 @@ public class CapFilter : SubscribeFilter
     /// <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.ContainsKey("cap-exception")) return Task.CompletedTask;
-
-        // 轨迹增改要做幂等
-        if (!context.DeliverMessage.Headers["cap-msg-name"].IsIn(GlobalConst.TrackUpsertSubscribe))
+        // 轨迹新增、更新要做幂等
+        if (!context.DeliverMessage.Headers["cap-msg-name"]
+                .IsIn(GlobalConst.TrackUpsertSubscribe, GlobalConst.TrackManualPushSubscribe))
             return Task.CompletedTask;
 
-        // 本地调试不做幂等
-        if (System.Diagnostics.Debugger.IsAttached)
-        {
-            return Task.CompletedTask;
-        }
-
         var argumentJsonStr = context.Arguments.ToJson();
         var hash = argumentJsonStr.GetHash();
 
@@ -44,15 +36,23 @@ public class CapFilter : SubscribeFilter
                 var isSkip = !string.IsNullOrEmpty(RedisHelper.Client.Get($"CapExecuteFilter:{hash}"));
                 if (isSkip)
                     throw new BusinessException(ErrorCode, $"检测到重复消费,消费参数：{argumentJsonStr}");
-
-                RedisHelper.Client.Set($"CapExecuteFilter:{hash}", 60 * 10,
-                    TimeSpan.FromMinutes(5).Seconds);
             }
         }
 
         return Task.CompletedTask;
     }
 
+    public override Task OnSubscribeExecutedAsync(ExecutedContext context)
+    {
+        context.DeliverMessage.Headers.TryGetValue("sjzy-arguments-hash", out var hash);
+        if (!hash.IsNullOrEmpty())
+        {
+            RedisHelper.Client.Set($"CapExecuteFilter:{hash}", 60 * 10, TimeSpan.FromMinutes(10).Seconds);
+        }
+
+        return Task.CompletedTask;
+    }
+
     /// <summary>
     /// 消费异常事件
     /// </summary>
@@ -72,4 +72,22 @@ public class CapFilter : SubscribeFilter
 
         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
 {
@@ -29,6 +29,8 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
     [CapSubscribe(GlobalConst.TrackDeleteSubscribe, Group = GlobalConst.TrackDeleteSubscribe)]
     public async Task ConsumerAsync(TrackDeleteEvent @event)
     {
+        throw new Exception();
+
         var dup = @event.Visited.Intersect(@event.OrderIds).ToArray();
         if (dup.Any())
         {

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
 {

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,6 +38,7 @@ 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);

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
 {

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/Services/ProvideServices/TrackPushService.cs

@@ -62,7 +62,7 @@ public class TrackPushService : ApplicationService
                 OperationMan = CurrentUserRequest.CurrentUser.Name,
             };
 
-            await _capPublisher.PublishAsync(GlobalConst.TrackPushSubscribe, trackPushEvent);
+            await _capPublisher.PublishAsync(GlobalConst.TrackManualPushSubscribe, trackPushEvent);
         }
 
         return $"即将推送轨迹代码{input.TrackCode}给{orders.Length}个订单";

src/SJZY.Track.Application/SjzyTrackApplicationModule.cs

@@ -1,12 +1,13 @@
 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,6 +75,7 @@ namespace SJZY.Track.Application
             );
 
             //CAP
+            context.Services.AddSingleton<ISubscribeExecutor, MySubscribeExecutor>();
             context.Services.AddCapTableNameService();
             context.Services.AddCap(options =>
             {

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>