不知何故，我认为流并不真正适合您尝试做的事情。您正在设置一种情况，即长期运行的程序将无缘无故地不断扩展数据需求。
我建议使用 pub/sub 模型将接收到的音频数据发布给订阅者，最好使用多线程方法来最大限度地减少不良订阅者的影响。一些想法可以在here找到。
我以前使用实现IObserver<byte[]> 并使用Queue<byte[]> 来存储样本块的处理器类完成此操作，直到进程线程为它们准备好。以下是基类：
public abstract class BufferedObserver<T> : IObserver<T>, IDisposable
{
private object _lck = new object();
private IDisposable _subscription = null;
public bool Subscribed { get { return _subscription != null; } }
private bool _completed = false;
public bool Completed { get { return _completed; } }
protected readonly Queue<T> _queue = new Queue<T>();
protected bool DataAvailable { get { lock(_lck) { return _queue.Any(); } } }
protected int AvailableCount { get { lock (_lck) { return _queue.Count; } } }
protected BufferedObserver()
{
}
protected BufferedObserver(IObservable<T> observable)
{
SubscribeTo(observable);
}
public virtual void Dispose()
{
if (_subscription != null)
{
_subscription.Dispose();
_subscription = null;
}
}
public void SubscribeTo(IObservable<T> observable)
{
if (_subscription != null)
_subscription.Dispose();
_subscription = observable.Subscribe(this);
_completed = false;
}
public virtual void OnCompleted()
{
_completed = true;
}
public virtual void OnError(Exception error)
{ }
public virtual void OnNext(T value)
{
lock (_lck)
_queue.Enqueue(value);
}
protected bool GetNext(ref T buffer)
{
lock (_lck)
{
if (!_queue.Any())
return false;
buffer = _queue.Dequeue();
return true;
}
}
protected T NextOrDefault()
{
T buffer = default(T);
GetNext(ref buffer);
return buffer;
}
}
public abstract class Processor<T> : BufferedObserver<T>
{
private object _lck = new object();
private Thread _thread = null;
private object _cancel_lck = new object();
private bool _cancel_requested = false;
private bool CancelRequested
{
get { lock(_cancel_lck) return _cancel_requested; }
set { lock(_cancel_lck) _cancel_requested = value; }
}
public bool Running { get { return _thread == null ? false : _thread.IsAlive; } }
public bool Finished { get { return _thread == null ? false : !_thread.IsAlive; } }
protected Processor(IObservable<T> observable)
: base(observable)
{ }
public override void Dispose()
{
if (_thread != null && _thread.IsAlive)
{
//CancelRequested = true;
_thread.Join(5000);
}
base.Dispose();
}
public bool Start()
{
if (_thread != null)
return false;
_thread = new Thread(threadfunc);
_thread.Start();
return true;
}
private void threadfunc()
{
while (!CancelRequested && (!Completed || _queue.Any()))
{
if (DataAvailable)
{
T data = NextOrDefault();
if (data != null && !data.Equals(default(T)))
ProcessData(data);
}
else
Thread.Sleep(10);
}
}
// implement this in a sub-class to process the blocks
protected abstract void ProcessData(T data);
}
这样，您只需在需要时保留数据，并且您可以将任意数量的进程线程附加到同一个可观察数据源。
为了完整起见，这里有一个实现IObservable<T> 的通用类，因此您可以看到它是如何组合在一起的。这个甚至有 cmets：
/// <summary>Generic IObservable implementation</summary>
/// <typeparam name="T">Type of messages being observed</typeparam>
public class Observable<T> : IObservable<T>
{
/// <summary>Subscription class to manage unsubscription of observers.</summary>
private class Subscription : IDisposable
{
/// <summary>Observer list that this subscription relates to</summary>
public readonly ConcurrentBag<IObserver<T>> _observers;
/// <summary>Observer to manage</summary>
public readonly IObserver<T> _observer;
/// <summary>Initialize subscription</summary>
/// <param name="observers">List of subscribed observers to unsubscribe from</param>
/// <param name="observer">Observer to manage</param>
public Subscription(ConcurrentBag<IObserver<T>> observers, IObserver<T> observer)
{
_observers = observers;
_observer = observer;
}
/// <summary>On disposal remove the subscriber from the subscription list</summary>
public void Dispose()
{
IObserver<T> observer;
if (_observers != null && _observers.Contains(_observer))
_observers.TryTake(out observer);
}
}
// list of subscribed observers
private readonly ConcurrentBag<IObserver<T>> _observers = new ConcurrentBag<IObserver<T>>();
/// <summary>Subscribe an observer to this observable</summary>
/// <param name="observer">Observer instance to subscribe</param>
/// <returns>A subscription object that unsubscribes on destruction</returns>
/// <remarks>Always returns a subscription. Ensure that previous subscriptions are disposed
/// before re-subscribing.</remarks>
public IDisposable Subscribe(IObserver<T> observer)
{
// only add observer if it doesn't already exist:
if (!_observers.Contains(observer))
_observers.Add(observer);
// ...but always return a new subscription.
return new Subscription(_observers, observer);
}
// delegate type for threaded invocation of IObserver.OnNext method
private delegate void delNext(T value);
/// <summary>Send <paramref name="data"/> to the OnNext methods of each subscriber</summary>
/// <param name="data">Data object to send to subscribers</param>
/// <remarks>Uses delegate.BeginInvoke to send out notifications asynchronously.</remarks>
public void Notify(T data)
{
foreach (var observer in _observers)
{
delNext handler = observer.OnNext;
handler.BeginInvoke(data, null, null);
}
}
// delegate type for asynchronous invocation of IObserver.OnComplete method
private delegate void delComplete();
/// <summary>Notify all subscribers that the observable has completed</summary>
/// <remarks>Uses delegate.BeginInvoke to send out notifications asynchronously.</remarks>
public void NotifyComplete()
{
foreach (var observer in _observers)
{
delComplete handler = observer.OnCompleted;
handler.BeginInvoke(null, null);
}
}
}
现在您可以创建一个Observable<byte[]> 用作您感兴趣的Process<byte[]> 实例的发送器。从输入流、音频阅读器等中提取数据块并将它们传递给 Notify 方法。只要确保你事先克隆了数组......