Блог

Я пытаюсь создать коллекцию, из keys которой .. если ключ не существует, тогда мне нужно DoSomeMethod() , а затем добавить ключ в коллекцию.

Проблема в том, что это должно иметь возможность обрабатывать несколько потоков, пытающихся добавить один и тот же ключ одновременно.

Если два потока имеют один и тот же ключ, то будет работать только один DoSomeMethod() , в то время как другой должен ждать.

Я рассматривал возможность использования ConcurrentDictionary метода and GetOrAdd (с параметром Func(..) param), но, похоже, оба они «запускаются» одновременно, если два потока имеют один и тот же ключ. Я думал, что реализация GetOrAdd будет

• блокировка ‘key’
• получить value из key .
• если нет value , то делайте что угодно .. и теперь установите значение.
• возврат value .
  … и любые другие key обращения будут ждать, пока не будет выполнена блокировка.

Похоже, что мой пользовательский метод, который GetOrAdd вызывает метод, не является потокобезопасным.

Документы MSDN также предполагают это?

Примечания
Если вы вызываете GetOrAdd одновременно в разных потоках, addValueFactory он может вызываться несколько раз, но его пара ключ / значение может не добавляться в словарь для каждого вызова

Пример покаяния: я копирую файлы из source в destination .

• При копировании source файла проверьте коллекцию, если мы пытались проверить и создать папку назначения.
• Если в коллекции нет key , проверьте, не существуют ли папки назначения… и создайте его, если он не существует.
• После создания целевой папки сохраните имя / путь к этой папке в коллекции.
• Повторите для всех файлов.

Таким образом, мы создаем папки назначения только в том случае, если мы этого еще не сделали.

Это похоже на то, что я хочу заблокировать КЛЮЧ коллекции….

 public class OnceOnlyConcurrent<TKey, TValue>
{
    private readonly ConcurrentDictionary<TKey, Lazy<TValue>> _dictionary = new ConcurrentDictionary<TKey, Lazy<TValue>>();

    public TValue GetOrAdd(TKey key, Func<TValue> computation)
    {
        var result = _dictionary.AddOrUpdate(key, _ => new Lazy<TValue>(computation, LazyThreadSafetyMode.ExecutionAndPublication), (_, v) => v);
        return result.Value;
    }
}
  

Думаю, я должен немного описать это. По сути, здесь происходит то, что, хотя AddOrUpdate addValueFactory делегат всегда вызывается дважды, если AddOrUpdate одновременно встречаются два вызывающих абонента, оба этих вызова на самом деле ничего не делают, кроме как возвращают Lazy<T> ссылку, которая завершает вычисление.

Внутри AddOrUpdate оба результата будут записаны, но один будет удален. Только один экземпляр Lazy<T> будет возвращен обоим вызывающим AddOrUpdate , поэтому Lazy<T> вызываемое вычисление будет определяться одним экземпляром.

Затем, в следующей строке, когда мы запрашиваем .Value , это фактически запустит вычисление для одного из вызывающих пользователей этого пользовательского GetOrAdd интерфейса, а другой заблокируется, пока первый вычисляет — это функциональность второго аргумента для Lazy<T> ( LazyThreadSafteMode.ExecutionAndPublication ) . Кстати, это поведение по умолчанию Lazy<T> , поэтому вам действительно не нужен второй аргумент — я просто использовал его, чтобы быть более понятным в этом сообщении.

Конечно, этот код также может быть написан как метод расширения, но, к сожалению, вам нужно знать, чтобы создать словарь с Lazy<T> объектами внутри, поэтому я думаю, что это лучше как класс-оболочка ConcurrentDictionary<TKey, TValue> .

В этом нет необходимости GetOrAdd . Достаточно простой проверки пути и наличия ключа:

 class FileWorker
{
    private object _sync;
    private IDictionary<string, Task> _destTasks;

    public FileWorker()
    {
        _sync = new object();
        _destTasks = new Dictionary<string, Task>();
    }

    public async Task Copy(IEnumerable<FileInfo> files, string destinationFolder)
    {
        await Task.WhenAll(files.Select(f => Copy(f, destinationFolder)));
    }

    private async Task CreateDestination(string path)
    {
        await Task.Run(() =>
        {
            if (!Directory.Exists(path))
            {
                Directory.CreateDirectory(path);
            }
        });
    }

    private Task Destination(string path)
    {
        lock(_sync)
        {
            if (!_destTasks.ContainsKey(path))
            {
                _destTasks[path] = CreateDestination(path);
            }
        }
        return _destTasks[path];
    }

    private async Task Copy(FileInfo file, string destinationFolder)
    {
        await Destination(destinationFolder).ContinueWith(task => file.CopyTo(Path.Combine(destinationFolder, file.Name), true));
    }
}

class Program
{
    static void Main(string[] args)
    {
        var file1 = new FileInfo("file1.tmp");
        using(var writer = file1.CreateText())
        {
            writer.WriteLine("file 1");
        }
        var file2 = new FileInfo("file2.tmp");
        using(var writer = file2.CreateText())
        {
            writer.WriteLine("file 2");
        }
        var worker = new FileWorker();
        worker.Copy(new[] { file1, file2 }, @"C:temp").Wait();
        Console.ReadLine();
    }
}