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

Question

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

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

1. 项目推送到队列
2. 项目从队列中弹出
3. 项目（未）从集合中移除
4. 项目已添加到集合中

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

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

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> 很容易地实现这一点，并锁定所有必要的位置。但是，我有兴趣使用 .NET 4.0 中的 ConcurrentDictionary<TKey, TValue> 和 ConcurrentQueue<T> 类（也可作为 .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

简而言之，问题中提供的代码不是线程安全的。

MSDN 文档中关于 AddOrUpdate 方法的内容相当少，因此我查看了 Reflector 中的 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方法中使用的完全一样。

注意代码如何使用ConcurrentDictionary 中的bool 值作为“虚拟”锁来同步队列和字典的协调。该数据结构还包含 CAS 等效操作TryUpdate，它被用来获取和释放这个“虚拟”锁。并且因为锁是“虚拟的”并且实际上并没有阻止并发访问，所以TryDequeue 方法中的while 循环是强制性的。这符合 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 方法具有 exact 相同的语句序列。

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

Answer 2

查看 Eric Lppert 的博客。你可能会找到你喜欢的东西... http://blogs.msdn.com/b/ericlippert/archive/tags/immutability/