c# – 非阻塞并发收集？

第一个不符合你的阻塞线程的要求,但我认为它相当优雅,因为你可以注册回调,它们将以循环方式被调用,但你仍然可以像往常一样调用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); // 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); // Transfer to the thread pool.
            }
            else
            {
                m_Callbacks.Enqueue(callback);
            }
        }
    }
}