c# - 在 MemoryCache 调用上停止重入

Question

应用程序需要加载数据并缓存一段时间。我希望如果应用程序的多个部分想要同时访问同一个缓存键，缓存应该足够智能，只加载一次数据并将该调用的结果返回给所有调用者。然而， MemoryCache 不是这样做的。如果您并行访问缓存(这通常发生在应用程序中)，它会为每次尝试获取缓存值创建一个任务。我认为这段代码会达到预期的结果，但事实并非如此。我希望缓存只运行一个 GetDataAsync任务，等待它完成，并使用结果获取其他调用的值。

using Microsoft.Extensions.Caching.Memory;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace ConsoleApp4
{
    class Program
    {
        private const string Key = "1";
        private static int number = 0;

        static async Task Main(string[] args)
        {
            var memoryCache = new MemoryCache(new MemoryCacheOptions { });

            var tasks = new List<Task>();
            tasks.Add(memoryCache.GetOrCreateAsync(Key, (cacheEntry) => GetDataAsync()));
            tasks.Add(memoryCache.GetOrCreateAsync(Key, (cacheEntry) => GetDataAsync()));
            tasks.Add(memoryCache.GetOrCreateAsync(Key, (cacheEntry) => GetDataAsync()));

            await Task.WhenAll(tasks);

            Console.WriteLine($"The cached value was: {memoryCache.Get(Key)}");
        }

        public static async Task<int> GetDataAsync()
        {
            //Simulate getting a large chunk of data from the database
            await Task.Delay(3000);
            number++;
            Console.WriteLine(number);
            return number;
        }
    }
}

这不是发生的事情。以上显示了这些结果(不一定按此顺序):

2

1

3

The cached value was: 3

它为每个缓存请求创建一个任务，并丢弃从其他两个请求返回的值。
这不必要地花费了时间，这让我想知道您是否可以说这个类甚至是线程安全的。 ConcurrentDictionary有相同的行为。我测试了它，同样的事情发生了。
有没有办法在任务不运行 3 次的情况下实现所需的行为？

Answer 1

这是自定义扩展方法 GetOrCreateExclusiveAsync ，类似于原生 IMemoryCache.GetOrCreateAsync ，这可以防止在正常条件下并发调用提供的异步 lambda。目的是在大量使用的情况下提高缓存机制的效率。仍然存在并发发生的可能性，因此这不能替代线程同步(如果需要)。
此实现还会从缓存中驱逐出错的任务，以便随后重试失败的异步操作。

using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.Primitives;

/// <summary>
/// Returns an entry from the cache, or creates a new cache entry using the
/// specified asynchronous factory method. Concurrent invocations are prevented,
/// unless the entry is evicted before the completion of the delegate. The errors
/// of failed invocations are not cached.
/// </summary>
public static Task<T> GetOrCreateExclusiveAsync<T>(this IMemoryCache cache, object key,
    Func<Task<T>> factory, MemoryCacheEntryOptions options = null)
{
    if (!cache.TryGetValue(key, out Task<T> task))
    {
        var entry = cache.CreateEntry(key);
        if (options != null) entry.SetOptions(options);
        var cts = new CancellationTokenSource();
        var newTaskTask = new Task<Task<T>>(async () =>
        {
            try { return await factory().ConfigureAwait(false); }
            catch { cts.Cancel(); throw; }
            finally { cts.Dispose(); }
        });
        var newTask = newTaskTask.Unwrap();
        entry.ExpirationTokens.Add(new CancellationChangeToken(cts.Token));
        entry.Value = newTask;
        entry.Dispose(); // The Dispose actually inserts the entry in the cache
        if (!cache.TryGetValue(key, out task)) task = newTask;
        if (task == newTask)
            newTaskTask.RunSynchronously(TaskScheduler.Default);
        else
            cts.Dispose();
    }
    return task;
}

用法示例:

var cache = new MemoryCache(new MemoryCacheOptions());
string html = await cache.GetOrCreateExclusiveAsync(url, async () =>
{
    return await httpClient.GetStringAsync(url);
}, new MemoryCacheEntryOptions().SetAbsoluteExpiration(TimeSpan.FromMinutes(10)));

这个实现在内部使用嵌套任务( Task<Task<T>> )而不是惰性任务( Lazy<Task<T>> )作为包装器，因为后面的构造在某些情况下容易发生死锁。
引用: Lazy<Task> with asynchronous initialization , VSTHRD011 Use AsyncLazy .

GitHub上相关API建议: GetOrCreateExclusive() and GetOrCreateExclusiveAsync(): Exclusive versions of GetOrCreate() and GetOrCreateAsync()