Merge pull request #471 from colinin/5.1.1

fix(idgenerator): 解决雪花算法时钟回退问题

aspnet-core/modules/common/LINGYUN.Abp.EventBus.CAP/LINGYUN/Abp/EventBus/CAP/AbpCAPSubscribeInvoker.cs

@@ -57,11 +57,12 @@ namespace LINGYUN.Abp.EventBus.CAP
             cancellationToken.ThrowIfCancellationRequested();
 
             var methodInfo = context.ConsumerDescriptor.MethodInfo;
-            var reflectedType = methodInfo.ReflectedType.Name;
+            var reflectedTypeHandle = methodInfo.ReflectedType!.TypeHandle.Value;
+            var methodHandle = methodInfo.MethodHandle.Value;
+            var key = $"{reflectedTypeHandle}_{methodHandle}";
 
             _logger.LogDebug("Executing subscriber method : {0}", methodInfo.Name);
 
-            var key = $"{methodInfo.Module.Name}_{reflectedType}_{methodInfo.MetadataToken}";
             var executor = _executors.GetOrAdd(key, x => ObjectMethodExecutor.Create(methodInfo, context.ConsumerDescriptor.ImplTypeInfo));
 
             using var scope = _serviceProvider.CreateScope();
@@ -77,9 +78,10 @@ namespace LINGYUN.Abp.EventBus.CAP
             var tenantId = message.GetTenantIdOrNull();
             for (var i = 0; i < parameterDescriptors.Count; i++)
             {
-                if (parameterDescriptors[i].IsFromCap)
+                var parameterDescriptor = parameterDescriptors[i];
+                if (parameterDescriptor.IsFromCap)
                 {
-                    executeParameters[i] = new CapHeader(message.Headers);
+                    executeParameters[i] = GetCapProvidedParameter(parameterDescriptor, message, cancellationToken);
                 }
                 else
                 {
@@ -87,7 +89,7 @@ namespace LINGYUN.Abp.EventBus.CAP
                     {
                         if (_serializer.IsJsonType(message.Value))  // use ISerializer when reading from storage, skip other objects if not Json
                         {
-                            var eventData = _serializer.Deserialize(message.Value, parameterDescriptors[i].ParameterType);
+                            var eventData = _serializer.Deserialize(message.Value, parameterDescriptor.ParameterType);
                             // 租户数据也可能存在事件数据中
                             if (tenantId == null && eventData is IMultiTenant tenant)
                             {
@@ -97,7 +99,7 @@ namespace LINGYUN.Abp.EventBus.CAP
                         }
                         else
                         {
-                            var converter = TypeDescriptor.GetConverter(parameterDescriptors[i].ParameterType);
+                            var converter = TypeDescriptor.GetConverter(parameterDescriptor.ParameterType);
                             if (converter.CanConvertFrom(message.Value.GetType()))
                             {
                                 var eventData = converter.ConvertFrom(message.Value);
@@ -110,7 +112,7 @@ namespace LINGYUN.Abp.EventBus.CAP
                             }
                             else
                             {
-                                if (parameterDescriptors[i].ParameterType.IsInstanceOfType(message.Value))
+                                if (parameterDescriptor.ParameterType.IsInstanceOfType(message.Value))
                                 {
                                     // 租户数据也可能存在事件数据中
                                     if (tenantId == null && message.Value is IMultiTenant tenant)
@@ -121,7 +123,7 @@ namespace LINGYUN.Abp.EventBus.CAP
                                 }
                                 else
                                 {
-                                    var eventData = Convert.ChangeType(message.Value, parameterDescriptors[i].ParameterType);
+                                    var eventData = Convert.ChangeType(message.Value, parameterDescriptor.ParameterType);
                                     // 租户数据也可能存在事件数据中
                                     if (tenantId == null && eventData is IMultiTenant tenant)
                                     {
@@ -185,6 +187,29 @@ namespace LINGYUN.Abp.EventBus.CAP
             }
         }
         /// <summary>
+        /// 
+        /// </summary>
+        /// <param name="parameterDescriptor"></param>
+        /// <param name="message"></param>
+        /// <param name="cancellationToken"></param>
+        /// <returns></returns>
+        /// <exception cref="ArgumentException"></exception>
+        private static object GetCapProvidedParameter(ParameterDescriptor parameterDescriptor, Message message,
+            CancellationToken cancellationToken)
+        {
+            if (typeof(CancellationToken).IsAssignableFrom(parameterDescriptor.ParameterType))
+            {
+                return cancellationToken;
+            }
+
+            if (parameterDescriptor.ParameterType.IsAssignableFrom(typeof(CapHeader)))
+            {
+                return new CapHeader(message.Headers);
+            }
+
+            throw new ArgumentException(parameterDescriptor.Name);
+        }
+        /// <summary>
         /// 获取事件处理类实例
         /// </summary>
         /// <param name="provider"></param>

aspnet-core/modules/common/LINGYUN.Abp.IdGenerator/LINGYUN/Abp/IdGenerator/Snowflake/SnowflakeIdGenerator.cs

@@ -20,8 +20,11 @@ namespace LINGYUN.Abp.IdGenerator.Snowflake
         protected int TimestampLeftShift { get; }
         protected long SequenceMask { get; }
 
+        protected SnowflakeIdOptions Options { get; }
+
         private SnowflakeIdGenerator(SnowflakeIdOptions options)
         {
+            Options = options;
             WorkerIdShift = options.SequenceBits;
             DatacenterIdShift = options.SequenceBits + options.WorkerIdBits;
             TimestampLeftShift = options.SequenceBits + options.WorkerIdBits + options.DatacenterIdBits;
@@ -82,15 +85,24 @@ namespace LINGYUN.Abp.IdGenerator.Snowflake
             {
                 var timestamp = TimeGen();
 
-                // TODO: 时间回退解决方案, 保存一个时间节点, 当服务器时间发生改变, 从保存的节点开始递增
                 if (timestamp < _lastTimestamp)
-                    throw new Exception(
-                        $"InvalidSystemClock: Clock moved backwards, Refusing to generate id for {_lastTimestamp - timestamp} milliseconds");
+                {
+                    // 如果启用此选项, 发生时间回退时使用上一个时间戳
+                    if (!Options.UsePreviousInTimeRollback)
+                    {
+                        throw new Exception(
+                            $"InvalidSystemClock: Clock moved backwards, Refusing to generate id for {_lastTimestamp - timestamp} milliseconds");
+                    }
+                    timestamp = _lastTimestamp;
+                }
 
                 if (_lastTimestamp == timestamp)
                 {
                     Sequence = (Sequence + 1) & SequenceMask;
-                    if (Sequence == 0) timestamp = TilNextMillis(_lastTimestamp);
+                    if (Sequence == 0L)
+                    {
+                        timestamp = TilNextMillis(_lastTimestamp);
+                    }
                 }
                 else
                 {
@@ -100,7 +112,8 @@ namespace LINGYUN.Abp.IdGenerator.Snowflake
                 _lastTimestamp = timestamp;
                 var id = ((timestamp - Twepoch) << TimestampLeftShift) |
                          (DatacenterId << DatacenterIdShift) |
-                         (WorkerId << WorkerIdShift) | Sequence;
+                         (WorkerId << WorkerIdShift) | 
+                         Sequence;
 
                 return id;
             }

aspnet-core/modules/common/LINGYUN.Abp.IdGenerator/LINGYUN/Abp/IdGenerator/Snowflake/SnowflakeIdOptions.cs

@@ -24,6 +24,10 @@ public class SnowflakeIdOptions
     public long Sequence { get; set; }
 
     public int SequenceBits { get; set; }
+    /// <summary>
+    /// 发生时间回退时使用上一个ID
+    /// </summary>
+    public bool UsePreviousInTimeRollback { get; set; }
 
     public SnowflakeIdOptions()
     {
@@ -31,5 +35,6 @@ public class SnowflakeIdOptions
         DatacenterIdBits = 5;
         Sequence = 0L;
         SequenceBits = 12;
+        UsePreviousInTimeRollback = true;
     }
 }

aspnet-core/modules/task-management/LINGYUN.Abp.BackgroundTasks.Abstractions/LINGYUN/Abp/BackgroundTasks/DefaultJobLockProvider.cs

@@ -1,4 +1,5 @@
-using System.Collections.Concurrent;
+using System;
+using System.Collections.Concurrent;
 using System.Threading;
 using System.Threading.Tasks;
 using Volo.Abp;
@@ -9,31 +10,47 @@ namespace LINGYUN.Abp.BackgroundTasks;
 [Dependency(TryRegister = true)]
 public class DefaultJobLockProvider : IJobLockProvider, ISingletonDependency
 {
-    private readonly ConcurrentDictionary<string, SemaphoreSlim> _localSyncObjects = new();
+    private readonly ConcurrentDictionary<string, JobLock> _localSyncObjects = new();
 
     public virtual Task<bool> TryLockAsync(string jobKey, int lockSeconds, CancellationToken cancellationToken = default)
     {
         Check.NotNullOrWhiteSpace(jobKey, nameof(jobKey));
-        if (_localSyncObjects.ContainsKey(jobKey))
+        if (_localSyncObjects.TryGetValue(jobKey, out var jobLock))
         {
-            return Task.FromResult(false);
+            if (jobLock.ExpirationTime > DateTime.UtcNow)
+            {
+                return Task.FromResult(false);
+            }
+            jobLock.ExpirationTime = DateTime.UtcNow.AddSeconds(lockSeconds);
+            return Task.FromResult(true);
         }
 
-        var semaphore = new SemaphoreSlim(1, 1);
-        return Task.FromResult(_localSyncObjects.TryAdd(jobKey, semaphore));
+        jobLock = new JobLock
+        {
+            ExpirationTime = DateTime.UtcNow.AddSeconds(lockSeconds),
+            Semaphore = new SemaphoreSlim(1, 1)
+        };
+
+        return Task.FromResult(_localSyncObjects.TryAdd(jobKey, jobLock));
     }
 
     public Task<bool> TryReleaseAsync(string jobKey, CancellationToken cancellationToken = default)
     {
         Check.NotNullOrWhiteSpace(jobKey, nameof(jobKey));
 
-        if (_localSyncObjects.TryGetValue(jobKey, out var semaphore))
+        if (_localSyncObjects.TryGetValue(jobKey, out var jobLock))
         {
-            semaphore.Dispose();
+            jobLock.Semaphore.Dispose();
 
             return Task.FromResult(_localSyncObjects.TryRemove(jobKey, out _));
         }
 
         return Task.FromResult(false);
     }
+
+    private class JobLock
+    {
+        public DateTime ExpirationTime { get; set; }
+        public SemaphoreSlim Semaphore { get; set; }
+    }
 }

aspnet-core/modules/task-management/LINGYUN.Abp.BackgroundTasks.Abstractions/LINGYUN/Abp/BackgroundTasks/JobRunnableContextExtensions.cs

@@ -42,9 +42,14 @@ public static class JobRunnableContextExtensions
     }
 
     public static bool TryGetMultiTenantId(this JobRunnableContext context, out Guid? tenantId)
+    {
+        return context.TryGetMultiTenantId(nameof(JobInfo.TenantId), out tenantId);
+    }
+
+    public static bool TryGetMultiTenantId(this JobRunnableContext context, string multiTenancyKey, out Guid? tenantId)
     {
         tenantId = null;
-        if (context.TryGetString(nameof(JobInfo.TenantId), out var tenantUUIdString) &&
+        if (context.TryGetString(multiTenancyKey, out var tenantUUIdString) &&
             Guid.TryParse(tenantUUIdString, out var tenantUUId))
         {
             tenantId = tenantUUId;

aspnet-core/modules/task-management/LINGYUN.Abp.BackgroundTasks.Jobs/LINGYUN/Abp/BackgroundTasks/Jobs/HttpRequestJob.cs

@@ -43,21 +43,21 @@ public class HttpRequestJob : IJobRunnable
     {
         var url = context.GetString(PropertyUrl);
         var method = context.GetString(PropertyMethod);
-        context.TryGetJobData(PropertyData, out var data);
-        context.TryGetJobData(PropertyContentType, out var contentType);
+        context.TryGetString(PropertyData, out var data);
+        context.TryGetString(PropertyContentType, out var contentType);
 
         var jsonSerializer = context.GetRequiredService<IJsonSerializer>();
 
         var httpRequestMesasge = new HttpRequestMessage(new HttpMethod(method), url);
-        if (data != null)
+        if (!data.IsNullOrWhiteSpace())
         {
             // TODO: 需要支持表单类型
 
             // application/json 支持
             httpRequestMesasge.Content = new StringContent(
-                jsonSerializer.Serialize(data),
+                data,
                 Encoding.UTF8,
-                contentType?.ToString() ?? MimeTypes.Application.Json);
+                contentType ?? MimeTypes.Application.Json);
         }
         if (context.TryGetJobData(PropertyHeaders, out var headers))
         {

aspnet-core/modules/task-management/LINGYUN.Abp.BackgroundTasks/LINGYUN/Abp/BackgroundTasks/Internal/InMemoryJobStore.cs

@@ -19,8 +19,12 @@ internal class InMemoryJobStore : IJobStore, ISingletonDependency
 
     public Task<List<JobInfo>> GetAllPeriodTasksAsync(CancellationToken cancellationToken = default)
     {
+        var status = new JobStatus[] { JobStatus.Running, JobStatus.FailedRetry };
+
         var jobs = _memoryJobStore
-            .Where(x => x.JobType == JobType.Period && x.Status == JobStatus.Running)
+            .Where(x => !x.IsAbandoned)
+            .Where(x => x.JobType == JobType.Period && status.Contains(x.Status))
+            .Where(x => (x.MaxCount == 0 || x.TriggerCount < x.MaxCount) || (x.MaxTryCount == 0 || x.TryCount < x.MaxTryCount))
             .OrderByDescending(x => x.Priority)
             .ToList();
 
@@ -30,8 +34,12 @@ internal class InMemoryJobStore : IJobStore, ISingletonDependency
     public Task<List<JobInfo>> GetWaitingListAsync(int maxResultCount, CancellationToken cancellationToken = default)
     {
         var now = DateTime.Now;
+        var status = new JobStatus[] { JobStatus.Running, JobStatus.FailedRetry };
+
         var jobs = _memoryJobStore
-            .Where(x => !x.IsAbandoned && x.JobType != JobType.Period && x.Status == JobStatus.Running)
+            .Where(x => !x.IsAbandoned)
+            .Where(x => x.JobType != JobType.Period && status.Contains(x.Status))
+            .Where(x => (x.MaxCount == 0 || x.TriggerCount < x.MaxCount) || (x.MaxTryCount == 0 || x.TryCount < x.MaxTryCount))
             .OrderByDescending(x => x.Priority)
             .ThenBy(x => x.TryCount)
             .ThenBy(x => x.NextRunTime)
@@ -75,8 +83,8 @@ internal class InMemoryJobStore : IJobStore, ISingletonDependency
     {
         var expiratime = DateTime.Now - jobExpiratime;
 
-        var expriaJobs = _memoryJobStore.Where(
-            x => x.Status == JobStatus.Completed &&
+        var expriaJobs = _memoryJobStore
+            .Where(x => x.Status == JobStatus.Completed &&
                 expiratime.CompareTo(x.LastRunTime ?? x.EndTime ?? x.CreationTime) <= 0)
             .Take(maxResultCount);