ConcurrentDictionary GetOrAdd async

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;

Метод GetOrAdd не очень подходит для этой цели. Поскольку он не гарантирует, что фабрика запускается только один раз, единственная цель, которую он имеет, - это небольшая оптимизация (незначительная, поскольку добавления в любом случае редки), поскольку ей не нужно дважды хешировать и находить правильный сегмент (что произойдет дважды, если вы получаете и устанавливаете с двумя отдельными вызовами).

Я бы посоветовал вам сначала проверить кеш, если вы не нашли значение в кеше, затем введите какую-либо форму критического раздела (блокировка, семафор и т. Д.), Повторно проверьте кеш, если все еще отсутствует, выберите значение и вставляем в кеш.

Это гарантирует, что ваш резервный магазин будет задействован только один раз; даже если несколько запросов получают промах кеша одновременно, только первый запрос фактически получит значение, другие запросы будут ожидать семафор, а затем вернутся раньше, поскольку они повторно проверяют кеш в критической секции.

Псевдо-код (с использованием SemaphoreSlim со счетчиком 1, так как вы можете ожидать его асинхронно):

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();
    }
}

Попробуйте этот метод расширения:

/// <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);
}

Вместо dict.GetOrAdd(key,key=>something(key)) вы используете await dict.GetOrAddAsync(key,async key=>await something(key)). Очевидно, что в этой ситуации вы просто пишете это как await dict.GetOrAddAsync(key,something), но я хотел прояснить это.

Что касается опасений по поводу сохранения порядка операций, у меня есть следующие наблюдения:

1. Использование обычного GetOrAdd даст тот же эффект, если вы посмотрите, как он реализован. Я буквально использовал тот же код и заставил его работать для async. Ссылка говорит

делегат valueFactory вызывается вне блокировок, чтобы избежать проблем, которые могут возникнуть при выполнении неизвестного кода под блокировкой. Следовательно, GetOrAdd не является атомарным по отношению ко всем другим операциям в классе ConcurrentDictionary ‹TKey, TValue›.

2. SyncRoot не поддерживается в ConcurrentDictionary, они используют внутренний механизм блокировки, поэтому блокировка невозможна. Однако использование собственного механизма блокировки работает только для этого метода расширения. Если вы используете другой поток (например, с помощью GetOrAdd), вы столкнетесь с той же проблемой.

Возможно, используется выделенный кеш памяти (например, new или старый MemoryCache классы или эту стороннюю библиотеку) предпочтительнее к использованию простого ConcurrentDictionary. Если вам действительно не нужны часто используемые функции, такие как истечение срока действия, сжатие на основе размера, автоматическое исключение записей, которые зависят от других записей, срок действия которых истек, или зависят от изменяемых внешних ресурсов (например, файлов, баз данных и т. Д.). Однако следует отметить, что _3 _ все еще может потребоваться некоторая работа для правильной обработки асинхронных делегатов, поскольку его нестандартное поведение не идеально.

Ниже приведен собственный метод расширения GetOrAddAsync для ConcurrentDictionary, которые имеют Task<TValue> значения. Он принимает фабричный метод и гарантирует, что метод будет вызван не более одного раза. Это также гарантирует, что неудачные задачи будут удалены из словаря. Эта реализация оптимизирована для случая, когда получение существующей задачи происходит часто, а создание новой - редко.

/// <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);
});

Для удаления проблемных задач эта реализация использует явно реализованный _ 9_. Более подробную информацию об этом API можно найти здесь. Начиная с .NET 5 и далее, новый person Theodor Zoulias    schedule 13.01.2021