如何在刷新缓存时提供陈旧数据？

Question

我处理多线程和缓存的标准方法是使用"Double-checked locking" 模式。在数据检索可能需要很长时间的情况下，这会导致后续线程在第一个线程刷新缓存时等待。如果请求的吞吐量比数据的新鲜度具有更高的优先级，我希望能够在刷新缓存时继续将陈旧的缓存数据提供给后续线程。

我在System.Runtime.Caching 中使用ObjectCache。放置在缓存中的项目有一个标志，指示数据是否过时。当项目过期并从缓存中删除时，我使用RemoveCallback 机制重新输入设置了过期标志的项目。

处理访问缓存的代码如下：

class Repository {
    static ObjectCache Cache = MemoryCache.Default;
    static readonly SemaphoreSlim RefreshCacheSemaphore = new SemaphoreSlim(1);

    static volatile bool DataIsBeingRefreshed;

    public async Task<object> GetData() {
        const string cacheKey = "Key";

        var cacheObject = Cache.Get(cacheKey) as CacheObject;

        if(cacheObject != null && (!cacheObject.IsStale || DataIsBeingRefreshed)) {
            return cacheObject.Item;
        }

        await RefreshCacheSemaphore.WaitAsync();

        try {
            // Check again that the cache item is still stale.
            cacheObject = Cache.Get(cacheKey) as CacheObject;

            if(cacheObject != null && !cacheObject.IsStale) {
                return cacheObject.Item;
            }

            DataIsBeingRefreshed = true;

            // Get data from database.
            // Store new data in cache.

            DataIsBeingRefreshed = false;

            // Return new data.
        } finally {
            RefreshCacheSemaphore.Release();
        }
    }
}

这样做的问题是，根据调用之间的时间，线程要么成功地提供陈旧数据，要么陷入等待输入受信号量保护的代码。理想情况下，我不希望任何线程在缓存刷新时等待。

或者，我可以将方法更改为：

public async Task<object> GetData() {
    const string cacheKey = "Key";

    var cacheObject = Cache.Get(cacheKey) as CacheObject;

    if(cacheObject != null && (!cacheObject.IsStale || DataIsBeingRefreshed)) {
        return cacheObject.Item;
    }

    // New semaphore.
    await GetStaleDataSemaphore.WaitAsync();

    try {
        cacheObject = Cache.Get(cacheKey) as CacheObject;

        if(cacheObject != null && DataIsBeingRefreshed) {
            return cacheObject.Item
        }

        DataIsBeingRefreshed = true;
    } finally {
        GetStaleDataSemaphore.Release();
    }

    await RefreshCacheSemaphore.WaitAsync();

    try {
        // Check again that the cache item is still stale.
        cacheObject = Cache.Get(cacheKey) as CacheObject;

        if(cacheObject != null && !cacheObject.IsStale)
        {
            return cacheObject.Item;
        }

        // Get data from database.
        // Store new data in cache.

        DataIsBeingRefreshed = false;

        // Return new data.
    }
    finally
    {
        RefreshCacheSemaphore.Release();
    }
}

这应该会减少等待刷新缓存的线程数量，但是，如果我错过了一些会导致没有线程等待的既定模式，我不想引入更多的锁定机制。

我是在正确的路线上还是有一个既定的模式来处理这个问题？

Answer 1

我无法理解自动从缓存中逐出项目并同时刷新这些项目的场景。

在以下缓存实现中，项目永远不会从缓存中删除，但您可能会扩展此方法以进行驱逐。

这个想法是，您可以为加载任务返回一个 Task，以防它是第一次加载（您不可能快速返回，因为此时没有任何价值），以及当有已经是一个可用的值。 该值将仅包含在 Task 和 Task.FromResult 中。

class Cache<TKey, TValue>
{
    private ConcurrentDictionary<TKey, Item> d = new ConcurrentDictionary<TKey, Item>();

    private class Item
    {
        public Item(Func<Task<TValue>> loadingTask, TimeSpan ttl, CancellationToken cancellationToken)
        {
            Ttl = ttl;
            LoadingTask = loadingTask;
            ServiceTask = HandleLoaded();
            CancellationToken = cancellationToken;
        }

        CancellationToken CancellationToken { get; }
        Func<Task<TValue>> LoadingTask { get; set; }
        public Task<TValue> ServiceTask {get; private set;}
        private TimeSpan Ttl { get; }

        async Task<TValue> HandleLoaded()
        {
            var value = await LoadingTask();
            ServiceTask = Task.FromResult(value);           
            Task.Run(() => ReloadOnExpiration(), CancellationToken);
            return value;
        }

        async void ReloadOnExpiration()
        {
            if (CancellationToken.IsCancellationRequested)
                return;
            await Task.Delay(Ttl, CancellationToken);
            var value = await LoadingTask();
            ServiceTask = Task.FromResult(value);
            ReloadOnExpiration();
        }
    }

    public async Task<TValue> GetOrCreate(TKey key, Func<TKey, CancellationToken, Task<TValue>> createNew, CancellationToken cancallationToken, TimeSpan ttl)
    {
        var item = d.GetOrAdd(key, k => new Item(() => createNew(key, cancallationToken), ttl, cancallationToken));
        return await item.ServiceTask;
    }
}

它的使用方法如下：

async Task Work(Cache<string, string> cache, CancellationToken cancellation)
{
    var item = await cache.GetOrCreate("apple", async (k, ct) =>
    {
        await Task.Delay(10, ct); // simulate loading time
        return $"{k} {Guid.NewGuid()}";
    }, cancellation, TimeSpan.FromMilliseconds(30));

    Console.WriteLine(item);
}