如何以线程安全的方式调用 .NET ConcurrentDictionary 上的 GetOrAdd 方法？

Question

我正在尝试拥有一个 keys 的集合，如果密钥不存在，那么我需要 DoSomeMethod() 然后将密钥添加到集合中。

问题是，这需要能够同时处理多个线程尝试添加相同的键。

如果两个线程有​​相同的键，那么只有一个会做DoSomeMethod()，而另一个需要等待。

我已经研究过使用ConcurrentDictionary 和GetOrAdd (with the Func(..) param option) 方法，但如果两个线程具有相同的密钥，这似乎会同时“启动”。我以为GetOrAdd 的实现会是

• 锁定“钥匙”
• 从key 获取value。
• 如果没有value 然后做任何事情.. 现在设置值。
• 返回value.
  ... 和任何其他 key 命中将等到锁定完成。

感觉就像GetOrAdd 方法调用的我的自定义方法不是线程安全的。

MSDN 文档也建议这样做？

备注
如果在不同线程上同时调用GetOrAdd，addValueFactory 可能会被多次调用，但它的键/值对可能不会每次调用都添加到字典中

---

悔过的例子：我正在将文件从source 复制到destination。

• 复制source 文件时，如果我们尝试检查并创建目标文件夹，请检查集合。
• 如果集合没有key，则检查目标文件夹是否不存在...如果不存在则创建它。
• 创建目标文件夹后，将此文件夹名称/路径存储在集合中。
• 对所有文件重复。

所以实际上，如果我们还没有创建目标文件夹，我们只会创建它。

这就像我想锁定一个集合 KEY ....

Answer 1

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

Answer 2

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>的包装类.