.net - 这似乎是并发集合/队列组合的合理方法吗？

Question

更新 :正如 Brian 指出的，我最初的想法确实存在并发问题。 ConcurrentDictionary<TKey, TValue>.AddOrUpdate 的签名有点掩盖了这一点。方法，它可以让懒惰的思考者(比如我自己)相信一切——集合添加和队列推送——会以某种方式以原子方式(即神奇地)同时发生。

回想起来，抱有这种期望是愚蠢的。其实不管AddOrUpdate的执行，应该清楚的是，在我最初的想法中仍然存在竞争条件，正如 Brian 指出的那样:推送到队列会在添加到集合之前发生，因此可能会发生以下事件序列:

项目插入队列

从队列中弹出的项目

项目(未)从集合中移除

添加到集合的项目

上述序列将导致集合中的一个项目不在队列中，从而有效地将该项目从数据结构中列入黑名单。

现在，我想了一会儿，我开始认为以下方法可以解决这些问题:

public bool Enqueue(T item)
{
    // This should:
    // 1. return true only when the item is first added to the set
    // 2. subsequently return false as long as the item is in the set;
    //    and it will not be removed until after it's popped
    if (_set.TryAdd(item, true))
    {
        _queue.Enqueue(item);
        return true;
    }

    return false;
}

以这种方式构建它， Enqueue调用只发生一次——在项目进入集合之后。所以队列中的重复项应该不是问题。并且似乎由于队列操作被集合操作“预定” - 即，仅在将项目添加到集合后才推送项目，并且在从集合中删除之前将其弹出 - 上面概述的有问题的事件序列不应该发生。

人们怎么想？难道这可以解决问题吗？ (就像布赖恩一样，我倾向于怀疑自己并猜测答案是否定的，我又错过了一些东西。但是，嘿，如果它很容易，那将不是一个有趣的挑战，对吧？)

我确实在 SO 上看到过类似的问题，但令人惊讶的是(考虑到这个网站的 .NET 繁重程度)它们似乎都是针对 Java 的。

我基本上需要一个线程安全的集合/队列组合类。换句话说，它应该是一个不允许重复的 FIFO 集合(因此如果相同的项目已经在队列中，后续的 Enqueue 调用将返回 false，直到项目从队列中弹出)。

我意识到我可以很容易地用一个简单的 HashSet<T> 来实现它。和 Queue<T>在所有必要的地方锁定。但是，我有兴趣使用 ConcurrentDictionary<TKey, TValue> 来完成它。和 ConcurrentQueue<T> 来自 .NET 4.0 的类(也可作为 .NET 3.5 的 Rx 扩展的一部分，这是我正在使用的)，我理解它是某种无锁集合*。

我的基本计划是实现这个集合，如下所示:

class ConcurrentSetQueue<T>
{
    ConcurrentQueue<T> _queue;
    ConcurrentDictionary<T, bool> _set;

    public ConcurrentSetQueue(IEqualityComparer<T> comparer)
    {
        _queue = new ConcurrentQueue<T>();
        _set = new ConcurrentDictionary<T, bool>(comparer);
    }

    public bool Enqueue(T item)
    {
        // This should:
        // 1. if the key is not present, enqueue the item and return true
        // 2. if the key is already present, do nothing and return false
        return _set.AddOrUpdate(item, EnqueueFirst, EnqueueSecond);
    }

    private bool EnqueueFirst(T item)
    {
        _queue.Enqueue(item);
        return true;
    }

    private bool EnqueueSecond(T item, bool dummyFlag)
    {
        return false;
    }

    public bool TryDequeue(out T item)
    {
        if (_queue.TryDequeue(out item))
        {
            // Another thread could come along here, attempt to enqueue, and
            // fail; however, this seems like an acceptable scenario since the
            // item shouldn't really be considered "popped" until it's been
            // removed from both the queue and the dictionary.
            bool flag;
            _set.TryRemove(item, out flag);

            return true;
        }

        return false;
    }
}

我是否正确地考虑了这一点？从表面上看，我在上面写的这个基本想法中看不到任何明显的错误。但也许我忽略了一些东西。或者也许使用 ConcurrentQueue<T>与 ConcurrentDictionary<T, bool>由于我没有想到的原因，这实际上不是一个明智的选择。或者也许其他人已经在某个经过实战验证的库中实现了这个想法，我应该使用它。

任何关于这个主题的想法或有用的信息将不胜感激!

*这是否严格准确，我不知道；但性能测试向我表明，它们的性能确实优于对许多消费者线程使用锁定的可比较的手动集合。

Answer 1

简而言之，问题中提供的代码不是线程安全的。
AddOrUpdate 上的 MSDN 文档相当稀疏。方法所以我看了一下AddOrUpdate反射器中的方法。这是基本算法(出于法律原因，我没有发布 Reflector 输出，而且它很容易自己做)。

TValue value;
do
{
  if (!TryGetValue(...))
  {
    value = AddValueFactoryDelegate(key);
    if (!TryAddInternal(...))
    {
      continue;
    }
    return value;
  }
  value = UpdateValueFactoryDelegate(key);
} 
while (!TryUpdate(...))
return value;

很明显 AddValueFactoryDelegate和 UpdateValueFactoryDelegate可以执行多次。此处无需进一步解释。应该很明显这将如何破坏您的代码。代表可以被执行多次，我实际上有点震惊。文档没有提到这一点。您会认为这将是非常重要的一点，因此调用者知道避免传递具有副作用的代表(就像您的情况一样)。

但是即使保证代表只执行一次，仍然会有问题。通过替换您的 Enqueue 应该很容易将问题序列可视化。使用 AddOrUpdate 内容的方法方法。 AddValueFactoryDelegate可以执行并将项目插入 _queue ，但在将项目添加到 _set 之前，线程可能会被上下文切换中断.然后第二个线程可以调用您的 TryDequeue方法并从 _queue 中提取该项目，但未能从 _set 中删除它因为它还没有在那里。

更新:

好吧，我认为它不可能让它工作。 ConcurrentQueue缺少一项关键操作。我相信你需要一个 CAS相当于 TryDequeue方法。如果存在这样的操作，那么我认为以下代码是正确的。我使用神话中的 TryDequeueCas方法，它接受一个比较值作为条件，当且仅当队列中的顶部项目等于比较值时，原子地执行此操作。这个想法与 Interlocked.CompareExchange 中使用的完全相同。方法。

注意代码如何使用 bool ConcurrentDictionary 中的值作为“虚拟”锁来同步队列和字典的协调。该数据结构还包含 CAS 等效操作 TryUpdate它被用来获取和释放这个“虚拟”锁。并且因为锁是“虚拟的”并且实际上不会阻塞并发访问， while在 TryDequeue 中循环方法是强制性的。这符合 CAS 操作的规范模式，因为它们通常在循环中执行，直到成功。

该代码还使用 .NET 4.0 style try-finally pattern用于锁定获取语义以帮助防范由带外(异步)异常引起的问题。

注意:同样，代码使用了神话般的 ConcurrentQueue.TryDequeueCas方法。

class ConcurrentSetQueue<T>
{
    ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
    ConcurrentDictionary<T, bool> _set = new ConcurrentDictionary<T, bool>();

    public ConcurrentSetQueue()
    {
    }

    public bool Enqueue(T item)
    {
        bool acquired = false;
        try
        {
            acquired = _set.TryAdd(item, true);
            if (acquired)
            {
                _queue.Enqueue(item);
                return true;
            }
            return false;
        }
        finally
        {
            if (acquired) _set.TryUpdate(item, false, true);
        }
    }

    public bool TryDequeue(out T item)
    {
        while (_queue.TryPeek(out item))
        {
            bool acquired = false;
            try
            {
                acquired = _set.TryUpdate(item, true, false);
                if (acquired)
                {
                    if (_queue.TryDequeueCas(out item, item))
                    {
                        return true;
                    }
                }
            }
            finally
            {
                if (acquired) _set.TryRemove(item, out acquired);
            }
        }
        item = default(T);
        return false;
    }
}

更新 2:

引用您的修改通知，与我的相比，它有多相似。事实上，如果你从我的变体中删除所有的绒毛， Enqueue方法具有完全相同的语句序列。

我更担心 TryDequeue不过，这就是为什么我添加了“虚拟”锁概念，这在我的实现中需要很多额外的东西。我特别担心访问数据结构的相反顺序(字典然后在 Enqueue 方法中排队，但在 TryDequeue 方法中排队然后字典)但是，我越想你修改的方法，我就越喜欢它。我现在认为是因为访问顺序相反，它是安全的!