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);
}
}
}
}
