From b5f7f866646fb0f64be1d62c24ed34a77e3e96e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9D=8E=E9=B9=8F=E5=86=9B?= <2357729423@qq.com> Date: Mon, 27 May 2024 11:51:34 +0800 Subject: [PATCH] =?UTF-8?q?opt=EF=BC=9A=E4=BC=98=E5=8C=96=E6=8E=A8?= =?UTF-8?q?=E9=80=81=E5=B9=82=E7=AD=89=EF=BC=8C=E4=BC=98=E5=8C=96cap?= =?UTF-8?q?=E9=87=8D=E8=AF=95=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/SJZY.Track.Api/appsettings.json | 2 +- .../{EventHandlers => CAP}/CAPFilter.cs | 54 ++-- ...TrackChangeAfterUpdateOrderEventHandler.cs | 6 +- .../EventHandlers/TrackDeleteEventHandler.cs | 4 +- .../TrackInsertForNotMappedEventHandler.cs | 2 +- .../TrackPushCompletedEventHandler.cs | 2 +- .../EventHandlers/TrackPushEventHandler.cs | 3 +- .../EventHandlers/TrackRefreshEventHandler.cs | 2 +- .../EventHandlers/TrackUpsertEventHandler.cs | 2 +- .../CAP/LoggerExtensions.cs | 89 ++++++ .../CAP/MySubscribeExecutor.cs | 268 ++++++++++++++++++ .../CAP/SubscriberExecutionFailedException.cs | 15 + .../ProvideServices/TrackPushService.cs | 2 +- .../SjzyTrackApplicationModule.cs | 4 +- .../Consts/GlobalConst.MQ.cs | 5 + 15 files changed, 431 insertions(+), 29 deletions(-) rename src/SJZY.Track.Application/{EventHandlers => CAP}/CAPFilter.cs (59%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs (92%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackDeleteEventHandler.cs (98%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackInsertForNotMappedEventHandler.cs (97%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackPushCompletedEventHandler.cs (98%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackPushEventHandler.cs (98%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackRefreshEventHandler.cs (99%) rename src/SJZY.Track.Application/{ => CAP}/EventHandlers/TrackUpsertEventHandler.cs (99%) create mode 100644 src/SJZY.Track.Application/CAP/LoggerExtensions.cs create mode 100644 src/SJZY.Track.Application/CAP/MySubscribeExecutor.cs create mode 100644 src/SJZY.Track.Application/CAP/SubscriberExecutionFailedException.cs diff --git a/src/SJZY.Track.Api/appsettings.json b/src/SJZY.Track.Api/appsettings.json index 676973d6..d69cb0c1 100644 --- a/src/SJZY.Track.Api/appsettings.json +++ b/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" ] } diff --git a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs b/src/SJZY.Track.Application/CAP/CAPFilter.cs similarity index 59% rename from src/SJZY.Track.Application/EventHandlers/CAPFilter.cs rename to src/SJZY.Track.Application/CAP/CAPFilter.cs index 1ce7c51b..a4e1f841 100644 --- a/src/SJZY.Track.Application/EventHandlers/CAPFilter.cs +++ b/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 /// 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; + } + /// /// 消费异常事件 /// @@ -72,4 +72,22 @@ public class CapFilter : SubscribeFilter return Task.CompletedTask; } + + + /// 获取字符串的hash值 + /// + /// + 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); + } } \ No newline at end of file diff --git a/src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs similarity index 92% rename from src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs index 3e3b1d07..c8f184ae 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackChangeAfterUpdateOrderEventHandler.cs +++ b/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(); 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 diff --git a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs similarity index 98% rename from src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs index bd85a916..fa2751bf 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackDeleteEventHandler.cs +++ b/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()) { diff --git a/src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackInsertForNotMappedEventHandler.cs similarity index 97% rename from src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackInsertForNotMappedEventHandler.cs index 9bd76c5e..b2f59331 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackInsertForNotMappedEventHandler.cs +++ b/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 { diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushCompletedEventHandler.cs similarity index 98% rename from src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackPushCompletedEventHandler.cs index a1024118..4658aa44 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushCompletedEventHandler.cs +++ b/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 { diff --git a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs similarity index 98% rename from src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs index 5edf8f4a..734246eb 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackPushEventHandler.cs +++ b/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); diff --git a/src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs similarity index 99% rename from src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackRefreshEventHandler.cs index 0e9fd334..f982343e 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackRefreshEventHandler.cs +++ b/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 { diff --git a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs b/src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs similarity index 99% rename from src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs rename to src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs index 201c4efc..46443ffb 100644 --- a/src/SJZY.Track.Application/EventHandlers/TrackUpsertEventHandler.cs +++ b/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 { diff --git a/src/SJZY.Track.Application/CAP/LoggerExtensions.cs b/src/SJZY.Track.Application/CAP/LoggerExtensions.cs new file mode 100644 index 00000000..8b430825 --- /dev/null +++ b/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}'."); + } +} \ No newline at end of file diff --git a/src/SJZY.Track.Application/CAP/MySubscribeExecutor.cs b/src/SJZY.Track.Application/CAP/MySubscribeExecutor.cs new file mode 100644 index 00000000..c9ff478b --- /dev/null +++ b/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 logger, + IOptions options, + IServiceProvider provider) + { + _provider = provider; + _logger = logger; + _options = options.Value; + + _dataStorage = _provider.GetRequiredService(); + Invoker = _provider.GetRequiredService(); + _hostName = Helper.GetInstanceHostname(); + } + + private ISubscribeInvoker Invoker { get; } + + public async Task ExecuteAsync(MediumMessage message, ConsumerExecutorDescriptor? descriptor = null, + CancellationToken cancellationToken = default) + { + if (descriptor == null) + { + var selector = _provider.GetRequiredService(); + 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 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 + { + [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() + .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 +} \ No newline at end of file diff --git a/src/SJZY.Track.Application/CAP/SubscriberExecutionFailedException.cs b/src/SJZY.Track.Application/CAP/SubscriberExecutionFailedException.cs new file mode 100644 index 00000000..31b183af --- /dev/null +++ b/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; +} \ No newline at end of file diff --git a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs b/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs index d4bcb72d..6897451e 100644 --- a/src/SJZY.Track.Application/Services/ProvideServices/TrackPushService.cs +++ b/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}个订单"; diff --git a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs b/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs index fa522c48..efa5aa41 100644 --- a/src/SJZY.Track.Application/SjzyTrackApplicationModule.cs +++ b/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(); context.Services.AddCapTableNameService(); context.Services.AddCap(options => { diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index 93fd1c53..378f0b89 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -32,6 +32,11 @@ public partial class GlobalConst /// public const string TrackPushSubscribe = "SJZY.Track.TrackPush"; + /// + /// 轨迹人工手动推送 + /// + public const string TrackManualPushSubscribe = "SJZY.Track.TrackManualPush"; + /// /// 轨迹推送完成（接口通回调） ///