fix: local lock/distributed lock

billhong-just · Sep 12, 2023 · b5749ed · b5749ed
1 parent 428f598
commit b5749ed
Showing 1 changed file with 56 additions and 3 deletions.
diff --git a/src/EasyCaching.Core/EasyCachingAbstractProvider.cs b/src/EasyCaching.Core/EasyCachingAbstractProvider.cs
@@ -366,9 +366,62 @@ public async Task<IDictionary<string, CacheValue<T>>> GetAllAsync<T>(IEnumerable
 
         public async Task<CacheValue<T>> GetAsync<T>(string cacheKey, Func<Task<T>> dataRetriever, Func<Task<TimeSpan>> expirationRetriever, CancellationToken cancellationToken = default)
         {
-            if (_lockFactory == null) return await BaseGetAsync<T>(cacheKey, dataRetriever, expirationRetriever, cancellationToken);
-            // TiSvc 沒用到，不實作
-            throw new NotImplementedException();
+            var operationId = s_diagnosticListener.WriteGetCacheBefore(new BeforeGetRequestEventData(CachingProviderType.ToString(), Name, nameof(GetAsync), new[] { cacheKey }));
+            Exception e = null;
+            try
+            {
+                if (_lockFactory == null) return await BaseGetAsync<T>(cacheKey, dataRetriever, expirationRetriever, cancellationToken);
+
+                var value = await BaseGetAsync<T>(cacheKey, cancellationToken);
+                if (value.HasValue) return value;
+
+                var @lock = _lockFactory.CreateLock(Name, $"{cacheKey}_Lock");
+                try
+                {
+                    if (!await @lock.LockAsync(_options.SleepMs, cancellationToken)) throw new TimeoutException();
+
+                    value = await BaseGetAsync<T>(cacheKey, cancellationToken);
+                    if (value.HasValue) return value;
+
+                    var task = dataRetriever();
+                    if (!task.IsCompleted &&
+                        await Task.WhenAny(task, Task.Delay(_options.LockMs, cancellationToken)) != task)
+                        throw new TimeoutException();
+
+                    var item = await task;
+                    if (item != null || _options.CacheNulls)
+                    {
+                        TimeSpan expiration = await expirationRetriever();
+                        await BaseSetAsync(cacheKey, item, expiration, cancellationToken);
+
+                        return new CacheValue<T>(item, true);
+                    }
+                    else
+                    {
+                        return CacheValue<T>.NoValue;
+                    }
+                }
+                finally
+                {
+                    await @lock.DisposeAsync();
+                }
+            }
+            catch (Exception ex)
+            {
+                e = ex;
+                throw;
+            }
+            finally
+            {
+                if (e != null)
+                {
+                    s_diagnosticListener.WriteGetCacheError(operationId, e);
+                }
+                else
+                {
+                    s_diagnosticListener.WriteGetCacheAfter(operationId);
+                }
+            }
         }
 
         public async Task<CacheValue<T>> GetAsync<T>(string cacheKey, Func<Task<T>> dataRetriever, TimeSpan expiration, CancellationToken cancellationToken = default)