如何实现 IObservable

Question

我希望能够将 Queue 用作 Observable，但我不知道如何创建它。我希望它能够在有人调用 Enqueue.

时推送数据

class Producer:IObservable<int>
{
    private object @lock = new object();
    private Queue<int> queue = new Queue<int>();
    List<IObserver<int>> observers = new List<IObserver<int>>();

    public Producer()
    {
    }
    public IObservable<int> ToObservable()
    {
        return ///
    }
    public bool Enqueue(int sample)
    {
        if (sample == null)
        {
            return false;
        }
        this.queue.Enqueue(sample);
        return true;
    }
    public int Dequeue()
    {
        if(!this.queue.TryDequeue(out Sample rez))
        {
            return 0;
        }
        return rez;
    }

    public IDisposable Subscribe(IObserver<int> observer)
    {
    }
}

我可以处理 Subscriber 和 IObserver 的部分我只是不知道如何将我的 Producer 对象包装在 IObservable

Answer 1

通常您不会自己实施 IObservable<T> 而选择 Reactive Extensions。将可观察对象视为 classic 事件的替代方案，如下所示：

class Producer
{
    private Queue<int> _queue = new Queue<int>();
    private Subject<int> _whenEnqueued = new Subject<int>();

    public IObservable<int> WhenEnqueued => _whenEnqueued.AsObservable();        

    public void Enqueue(int value)
    {
        _queue.Enqueue(value);
        _whenEnqueued.OnNext(value);
    }
}

这样其他 classes 可以订阅 WhenEnqueued 并通过将工作委托给 Subject class 来获得新值的通知。

如果您真的想实现 IObservable<T>（通常您不会），请查看 IObservable documentation

中的示例

如何实现 IObservable

How to implement an IObservable

c#

system.reactive

observable

.net-core