ConcurrentDictionary GetOrAdd 异步

Question

我想使用GetOrAdd 和ConcurrentDictionary 之类的东西作为Web 服务的缓存。这本词典有异步版本吗？ GetOrAdd 将使用 HttpClient 发出 Web 请求，所以如果此字典有一个版本，其中 GetOrAdd 是异步的，那就太好了。

为了消除一些混乱，字典的内容将是对网络服务的调用的响应。

ConcurrentDictionary<string, Response> _cache
    = new ConcurrentDictionary<string, Response>();

var response = _cache.GetOrAdd("id",
    (x) => { _httpClient.GetAsync(x).GetAwaiter().GetResponse(); });

Answer 1

GetOrAdd 不会成为异步操作，因为访问字典的值不是长时间运行的操作。

但是，您可以做的只是将任务存储在字典中，而不是物化结果。任何需要结果的人都可以等待该任务。

但是，您还需要确保该操作只启动一次，而不是多次。为了保证某些操作只运行一次，而不是多次运行，还需要添加Lazy：

ConcurrentDictionary<string, Lazy<Task<Response>>> _cache = new ConcurrentDictionary<string, Lazy<Task<Response>>>();

var response = await _cache.GetOrAdd("id", url => new Lazy<Task<Response>>(_httpClient.GetAsync(url))).Value;

Answer 2

GetOrAdd 方法不太适合用于此目的。由于它不能保证工厂只运行一次，它的唯一目的是进行次要优化（次要，因为无论如何添加都很少），因为它不需要哈希并找到正确的存储桶两次（如果你得到和设置两个单独的调用）。

我建议你先检查缓存，如果在缓存中没有找到值，然后进入某种形式的临界区（锁、信号量等），重新检查缓存，如果仍然丢失则获取值并插入缓存。

这可确保您的后备存储只被命中一次；即使多个请求同时发生缓存未命中，也只有第一个会真正获取值，其他请求将等待信号量然后提前返回，因为它们会重新检查临界区的缓存。

伪代码（使用计数为 1 的 SemaphoreSlim，因为您可以异步等待它）：

async Task<TResult> GetAsync(TKey key)
{
    // Try to fetch from catch
    if (cache.TryGetValue(key, out var result)) return result;

    // Get some resource lock here, for example use SemaphoreSlim 
    // which has async wait function:
    await semaphore.WaitAsync();    
    try 
    {
        // Try to fetch from cache again now that we have entered 
        // the critical section
        if (cache.TryGetValue(key, out result)) return result;

        // Fetch data from source (using your HttpClient or whatever), 
        // update your cache and return.
        return cache[key] = await FetchFromSourceAsync(...);
    }
    finally
    {
        semaphore.Release();
    }
}

Answer 3

试试这个扩展方法：

/// <summary>
/// Adds a key/value pair to the <see cref="ConcurrentDictionary{TKey, TValue}"/> by using the specified function 
/// if the key does not already exist. Returns the new value, or the existing value if the key exists.
/// </summary>
public static async Task<TResult> GetOrAddAsync<TKey,TResult>(
    this ConcurrentDictionary<TKey,TResult> dict,
    TKey key, Func<TKey,Task<TResult>> asyncValueFactory)
{
    if (dict.TryGetValue(key, out TResult resultingValue))
    {
        return resultingValue;
    }
    var newValue = await asyncValueFactory(key);
    return dict.GetOrAdd(key, newValue);
}

您使用await dict.GetOrAddAsync(key,async key=>await something(key))，而不是dict.GetOrAdd(key,key=>something(key))。显然，在这种情况下，您只需将其写为await dict.GetOrAddAsync(key,something)，但我想说清楚。

关于保持操作顺序的问题，我有以下观察：

1. 如果您查看它的实现方式，使用普通的 GetOrAdd 将获得相同的效果。我确实使用了相同的代码并使其适用于异步。参考资料说

valueFactory 委托在锁之外被调用以避免 在锁定下执行未知代码可能引起的问题。 因此，GetOrAdd 对于所有其他操作都不是原子的 在 ConcurrentDictionary 类

1. ConcurrentDictionary 不支持 SyncRoot，它们使用内部锁定机制，因此无法对其进行锁定。但是，使用您自己的锁定机制仅适用于此扩展方法。如果您使用其他流程（例如使用 GetOrAdd），您将面临同样的问题。

Answer 4

可能使用专用内存缓存（如new 或old MemoryCache 类，或this 第三方库）应该比使用简单的ConcurrentDictionary 更好。除非你真的不需要常用的功能，例如基于时间的过期、基于大小的压缩、依赖于其他已过期条目或依赖于可变外部资源（如文件、数据库等）的条目的自动驱逐。应该注意的是，MemoryCache 可能仍需要一些工作才能正确处理异步委托，因为它的开箱即用行为is not ideal。

下面是具有Task<TValue> 值的ConcurrentDictionarys 的自定义扩展方法GetOrAddAsync。它接受一个工厂方法，并确保该方法最多被调用一次。它还确保从字典中删除失败的任务。此实现针对频繁获取现有任务而很少发生创建新任务的情况进行了优化。

/// <summary>
/// Returns an existing task from the concurrent dictionary, or adds a new task
/// using the specified asynchronous factory method. Concurrent invocations for
/// the same key are prevented, unless the task is removed before the completion
/// of the delegate. Failed tasks are evicted from the concurrent dictionary.
/// </summary>
public static Task<TValue> GetOrAddAsync<TKey, TValue>(
    this ConcurrentDictionary<TKey, Task<TValue>> source, TKey key,
    Func<TKey, Task<TValue>> valueFactory)
{
    if (!source.TryGetValue(key, out var currentTask))
    {
        Task<TValue> newTask = null;
        var newTaskTask = new Task<Task<TValue>>(async () =>
        {
            try { return await valueFactory(key).ConfigureAwait(false); }
            catch
            {
                ((ICollection<KeyValuePair<TKey, Task<TValue>>>)source)
                    .Remove(new KeyValuePair<TKey, Task<TValue>>(key, newTask));
                //source.TryRemove(KeyValuePair.Create(key, newTask)); // .NET 5
                throw;
            }
        });
        newTask = newTaskTask.Unwrap();
        currentTask = source.GetOrAdd(key, newTask);
        if (currentTask == newTask)
            newTaskTask.RunSynchronously(TaskScheduler.Default);
    }
    return currentTask;
}

使用示例：

var cache = new ConcurrentDictionary<string, Task<HttpResponseMessage>>();

var response = await cache.GetOrAddAsync("https://stackoverflow.com", async url =>
{
    return await _httpClient.GetAsync(url);
});

为了移除故障任务，此实现使用显式实现的ICollection<T>.Remove API。有关此 API 的更多信息，请访问 here。从 .NET 5 开始，可以改用新的 TryRemove(KeyValuePair<TKey, TValue> item) 方法。

顺便说一句，如果需要最高性能，您可能需要查看BitFaster.Caching 第三方库。我从未亲自使用过它，但带有基准的图表看起来令人印象深刻。

Answer 5

我早在ConcurrentDictionary 和TPL 诞生之前就解决了这个问题。我在一家咖啡馆，没有那个原始代码，但它是这样的。

这不是一个严格的答案，但可能会激发您自己的解决方案。重要的是返回刚刚添加的值或已经存在的值以及布尔值，以便您可以分叉执行。

该设计让您可以轻松地将比赛获胜逻辑与失败逻辑分叉。

public bool TryAddValue(TKey key, TValue value, out TValue contains)
{
    // guards etc.

    while (true)
    {
        if (this.concurrentDic.TryAdd(key, value))
        {
            contains = value;
            return true;
        }
        else if (this.concurrentDic.TryGetValue(key, out var existing))
        {
            contains = existing;
            return false;
        }
        else
        {
            // Slipped down the rare path. The value was removed between the
            // above checks. I think just keep trying because we must have
            // been really unlucky.

            // Note this spinning will cause adds to execute out of
            // order since a very unlucky add on a fast moving collection
            // could in theory be bumped again and again before getting
            // lucky and getting its value added, or locating existing.

            // A tiny random sleep might work. Experiment under load.
        }
    }
}

这可以作为ConcurrentDictionary 的扩展，或者是你自己的缓存或使用锁的东西。

也许GetOrAdd(K,V) 可以与Object.ReferenceEquals() 一起使用来检查它是否被添加，而不是旋转设计。

说实话，上面的代码不是我回答的重点。强大之处在于方法签名的简单设计以及它如何提供以下功能：

static readonly ConcurrentDictionary<string, Task<Task<Thing>>> tasks = new();

//

var newTask = new Task<Task<Thing>>(() => GetThingAsync(thingId));

if (this.tasks.TryAddValue(thingId, newTask, out var task))
{
    task.Start();
}

var thingTask = await task;
var thing = await thingTask;

Task 需要如何保存Task 有点奇怪（如果您的工作是异步的），还有未使用的Tasks 的分配需要考虑。

我认为微软没有使用这种方法发布其线程安全集合或提取“并发集合”接口是一种耻辱。

我真正的实现是一个带有复杂的过期内部集合和东西的缓存。我想你可以继承 .NET Task 类并添加一个 CreatedAt 属性来帮助驱逐。

免责声明我根本没有尝试过这个，这太离谱了，但我在 2009 年在一个超高吞吐量的应用程序中使用了这种设计。