c# - 将两个 Observable 与一个优先级更高的 Observable 合并

Question

是否可以使用 ReactiveExtensions 来实现以下目标；

• 两个 Observable，一个是“高”优先级，另一个是“低”优先级

• 将两个 Observable 合并为一个，然后可以对其进行订阅，目的是让这个生成的 Observable 始终先于任何低优先级的项目发出高优先级的项目。

我知道这可以使用两个 ConcurrentQueue 集合和类似的东西更简单地实现；

return this.highPriorityItems.TryDequeue(out item) 
    || this.lowPriorityItems.TryDequeue(out item);

但是这种方法有一些问题，比如不能像 Observable 那样“可订阅”(所以一旦队列用完，处理就会结束，没有很多额外的废话来把它推到任务中)。

此外，我对在队列上应用一些额外的过滤很感兴趣，例如节流和“在更改之前不同”，因此 Rx 似乎很适合这里。

Answer 1

你说的当然是优先队列。

Rx 是关于事件的流，而不是队列。当然，队列 在 Rx 中被大量使用 - 但它们不是一流的概念，更多是 Rx 概念的实现细节的一部分。

我们需要队列的一个很好的例子是处理一个缓慢的观察者。事件在 Rx 中按顺序调度，如果事件到达的速度快于观察者可以处理的速度，那么它们必须针对该观察者排队。如果有很多观察者，则必须维护多个逻辑队列，因为观察者可能以不同的速度前进 - 并且 Rx 选择不让它们保持同步。

“背压”是观察者向可观察对象提供反馈的概念，以便允许机制处理更快可观察对象的压力——例如合并或节流。 Rx 没有引入背压的一流方法 - 唯一内置的方法是通过 OnNext 的同步特性来监视观察者。任何其他机制都需要带外。您的问题与背压直接相关，因为它仅在观察者速度慢的情况下才相关。

我提到所有这些是为了证明我的说法，即 Rx 不是提供您正在寻找的那种优先级调度的好选择 - 实际上，一流的排队机制似乎更合适。

要解决手头的问题，您需要在自定义运算符中自行管理优先级排队。重述这个问题:你的意思是，如果事件在观察者处理 OnNext 事件期间到达，那么就会有大量事件要分派(dispatch)，而不是典型的 FIFO 队列Rx 使用的那个，你想根据一些优先级进行调度。

需要注意的是，本着 Rx 如何不让多个观察者保持同步的精神，并发观察者可能会以不同的顺序看到事件，这对您来说可能是也可能不是问题。您可以使用 Publish 之类的机制来获得顺序一致性 - 但您可能不想这样做，因为在这种情况下事件传递的时间会变得非常不可预测且效率低下。

我确信有更好的方法可以做到这一点，但这里有一个基于优先级队列的交付示例 - 您可以使用更好的方法将其扩展为适用于多个流和优先级(甚至每个事件的优先级)队列实现(例如基于 b 树的优先级队列)，但我选择保持它相当简单。即便如此，请注意代码必须解决的大量问题，围绕错误处理、完成等 - 我已经选择了何时发出这些信号，肯定还有很多其他有效的选择。

总而言之，这个实现肯定会让我放弃为此使用 Rx 的想法。它足够复杂，无论如何这里都可能存在错误。正如我所说，可能会有更简洁的代码(特别是考虑到我已经付出了最少的努力!)，但概念上，无论实现如何，我都对这个想法感到不舒服:

public static class ObservableExtensions
{
    public static IObservable<TSource> MergeWithLowPriorityStream<TSource>(
        this IObservable<TSource> source,
        IObservable<TSource> lowPriority,
        IScheduler scheduler = null)
    {    
        scheduler = scheduler ?? Scheduler.Default;
        return Observable.Create<TSource>(o => {    
            // BufferBlock from TPL dataflow is used as it is
            // handily awaitable. package: Microsoft.Tpl.Dataflow        
            var loQueue = new BufferBlock<TSource>();
            var hiQueue = new BufferBlock<TSource>();
            var errorQueue = new BufferBlock<Exception>();
            var done = new TaskCompletionSource<int>();
            int doneCount = 0;
            Action incDone = () => {
                var dc = Interlocked.Increment(ref doneCount);
                if(dc == 2)
                    done.SetResult(0);
            };
            source.Subscribe(
                x => hiQueue.Post(x),
                e => errorQueue.Post(e),
                incDone);
            lowPriority.Subscribe(
                x => loQueue.Post(x),
                e => errorQueue.Post(e),
                incDone);
            return scheduler.ScheduleAsync(async(ctrl, ct) => {
                while(!ct.IsCancellationRequested)
                {
                    TSource nextItem;
                    if(hiQueue.TryReceive(out nextItem)
                      || loQueue.TryReceive(out nextItem))
                        o.OnNext(nextItem);

                    else if(done.Task.IsCompleted)
                    {
                        o.OnCompleted();
                        return;
                    }

                    Exception error;                        
                    if(errorQueue.TryReceive(out error))
                    {
                        o.OnError(error);
                        return;
                    }

                    var hiAvailableAsync = hiQueue.OutputAvailableAsync(ct);    
                    var loAvailableAsync = loQueue.OutputAvailableAsync(ct);                    
                    var errAvailableAsync =
                        errorQueue.OutputAvailableAsync(ct);
                    await Task.WhenAny(
                        hiAvailableAsync,
                        loAvailableAsync,
                        errAvailableAsync,
                        done.Task);
                }
            });
        });
    }
}

和示例用法:

void static Main()
{
    var xs = Observable.Range(0, 3);
    var ys = Observable.Range(10, 3);

    var source = ys.MergeWithLowPriorityStream(xs);

    source.Subscribe(Console.WriteLine, () => Console.WriteLine("Done"));
}

这将首先打印出 ys 的元素，表明它们的优先级更高。