c# – 非阻塞并发收集？

System.Collections.Concurrent有一些在多线程环境中运行良好的新集合.但是,它们有点受限.它们会阻塞直到某个项目可用,或者它们返回默认值(T)(TryXXX方法).

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

我目前的解决方案是使用BlockingCollection,但要将APM与委托一起使用以获取下一个元素.换句话说,我创建了一个从集合中获取的方法的委托,并使用BeginInvoke执行该委托.

不幸的是,为了实现这一目标,我必须在课堂上保持很多状态.更糟糕的是,这个类不是线程安全的;它只能由单个线程使用.我正在避开可维护性的边缘,我不想这样做.

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

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

TL;博士：

是否有任何满足要求的模式：

“我需要发信号通知我已准备好接收下一个元素的集合,并让集合在下一个元素到达时执行回调,而不会阻塞任何线程.”

解决方法

我想我有两种可能的解决方案.我对这两者都不是特别满意,但他们至少提供了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); // 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);
            }
        }
    }
}