c# - 非阻塞并发收集？

Question

System.Collections.Concurrent 有一些在多线程环境中工作得很好的新集合。但是，它们有点受限。它们要么阻塞直到项目可用，要么返回 default(T)(TryXXX 方法)。

我需要一个线程安全的集合，但它没有阻塞调用线程，而是使用回调来通知我至少有一个项目可用。

我当前的解决方案是使用 BlockingCollection，但使用带有委托(delegate)的 APM 来获取下一个元素。换句话说，我创建了一个方法的委托(delegate)，该方法从集合中获取 Take，并使用 BeginInvoke 执行该委托(delegate)。

不幸的是，为了实现这一点，我必须在类(class)中保持很多状态。更糟糕的是，这个类不是线程安全的；它只能由单个线程使用。我正在绕过可维护性的边缘，我不想这样做。

我知道有一些库可以让我在这里做的事情变得非常简单(我相信 Reactive Framework 就是其中之一)，但我想在不添加任何版本 4 之外的引用的情况下实现我的目标框架。

有没有我可以使用的更好的模式，不需要外部引用来实现我的目标？

---

tl;博士:

是否有满足要求的模式:

“我需要向集合发出信号，表明我已为下一个元素做好准备，并让集合在下一个元素到达时执行回调，而不会阻塞任何线程。”

Answer 1

我想我有两个可能的解决方案。我对两者都不是特别满意，但它们至少提供了 APM 方法的合理替代方案。

第一个不满足你的无阻塞线程的要求，但我认为它相当优雅，因为你可以注册回调，它们将以循环方式被调用，但你仍然可以调用 Take 或 TryTake，就像您通常对 BlockingCollection 所做的那样。每次请求项目时，此代码都会强制注册回调。这就是集合的信号机制。这种方法的好处是，对 Take 的调用不会像在我的第二个解决方案中那样被饿死。

public class NotifyingBlockingCollection<T> : BlockingCollection<T>
{
    private Thread m_Notifier;
    private BlockingCollection<Action<T>> m_Callbacks = new BlockingCollection<Action<T>>();

    public NotifyingBlockingCollection()
    {
        m_Notifier = new Thread(Notify);
        m_Notifier.IsBackground = true;
        m_Notifier.Start();
    }

    private void Notify()
    {
        while (true)
        {
            Action<T> callback = m_Callbacks.Take();
            T item = Take();
            callback.BeginInvoke(item, null, null); // Transfer to the thread pool.
        }
    }

    public void RegisterForTake(Action<T> callback)
    {
        m_Callbacks.Add(callback);
    }
}

第二个确实满足了无阻塞线程的要求。请注意它如何将回调的调用传输到线程池。我这样做是因为我认为如果它被同步执行，那么锁将保持更长时间，从而导致 Add 和 RegisterForTake 的瓶颈。我已经仔细查看了它，我认为它不会被实时锁定(一个项目和一个回调都可用，但回调永远不会被执行)但你可能想自己查看一下以进行验证。这里唯一的问题是对 Take 的调用会被饿死，因为回调总是优先。

public class NotifyingBlockingCollection<T>
{
    private BlockingCollection<T> m_Items = new BlockingCollection<T>();
    private Queue<Action<T>> m_Callbacks = new Queue<Action<T>>();

    public NotifyingBlockingCollection()
    {
    }

    public void Add(T item)
    {
        lock (m_Callbacks)
        {
            if (m_Callbacks.Count > 0)
            {
                Action<T> callback = m_Callbacks.Dequeue();
                callback.BeginInvoke(item, null, null); // Transfer to the thread pool.
            }
            else
            {
                m_Items.Add(item);
            }
        }
    }

    public T Take()
    {
        return m_Items.Take();
    }

    public void RegisterForTake(Action<T> callback)
    {
        lock (m_Callbacks)
        {
            T item;
            if (m_Items.TryTake(out item))
            {
                callback.BeginInvoke(item, null, null); // Transfer to the thread pool.
            }
            else
            {
                m_Callbacks.Enqueue(callback);
            }
        }
    }
}