您所描述的当然是优先级队列。
Rx 是关于事件的流，而不是队列。当然，队列在 Rx 中被大量使用 - 但它们不是一流的概念，更多的是 Rx 概念的实现细节。
我们需要队列的一个很好的例子是处理一个缓慢的观察者。事件在 Rx 中按顺序分派，如果事件到达的速度比观察者处理它们的速度快，那么它们必须针对该观察者排队。如果有很多观察者，则必须维护多个逻辑队列，因为观察者可能会以不同的速度前进 - 而 Rx 选择不让它们保持同步。
“背压”是观察者向可观察对象提供反馈的概念，以允许机制处理更快的可观察对象的压力 - 例如合并或节流。 Rx 没有一流的引入背压的方法——可观察者监控观察者的唯一内置方法是通过OnNext 的同步特性。任何其他机制都需要带外。您的问题与背压直接相关，因为它仅在观察者缓慢的情况下才相关。
我提到这一切是为了证明我的主张，即 Rx 不是提供您正在寻找的那种优先级调度的好选择 - 实际上，一流的排队机制似乎更合适。
要解决手头的问题，您需要在自定义运算符中自己管理优先级排队。重申问题：您的意思是，如果事件在观察者处理OnNext 事件期间到达，这样就会有大量要调度的事件，而不是 Rx 使用的典型 FIFO 队列，您想要根据某个优先级进行调度。
需要注意的是，本着 Rx 不让多个观察者保持同步的精神，并发观察者可能会以不同的顺序查看事件，这对您来说可能是也可能不是问题。您可以使用Publish 之类的机制来获得订单一致性 - 但您可能不想这样做，因为在这种情况下，事件传递的时间会变得非常不可预测且效率低下。
我确信有更好的方法可以做到这一点，但这里有一个基于优先级队列的交付示例 - 您可以使用更好的方法将其扩展为适用于多个流和优先级（甚至是每个事件的优先级）队列实现（例如基于 b 树的优先级队列），但我选择保持这个相当简单。即便如此，请注意代码必须解决的大量问题，包括错误处理、完成等。我已经做出了选择，当这些问题发出信号时，肯定还有很多其他有效的选择。
总而言之，这个实现确实让 me 放弃了使用 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的元素，表明它们的优先级更高。