c# - 同步多个 observable 的顺序

Question

我有多个(2-10)个热可观察对象，每个都在给定的时间窗口内触发一个事件，但没有按顺序触发。

假设我有 3 个对象，为了说明它们按以下任意顺序开火:

Subjects: sA, sB, sC

Order of time window 1:
  sB, then sA, then sC
Order of time window 2:
  sA, then sC, then sB
Order of time window 3:
  sA, then sB, then sC
etc.

因此，如果我进行合并，我将得到以下流:sB、sA、sC、sA、sC、sB、sA、sB、sC 等。

现在，这是我的问题:我想强制每个时间窗口内的事件顺序。

1. sA
2. sB
3. sC

这将导致以下合并流:sA、sB、sC、sA、sB、sC、sA、sB、sC 等。

窗口时间本身是未知的，但由于每个事件在每个窗口中恰好触发一次，我们可以假设一旦所有对象都触发了窗口，窗口就会关闭并打开一个新窗口。

任何想法如何优雅地做到这一点？

扩展问题(不那么重要):observables 是相互独立的，所以我需要引入一个通用的静态方法，我可以在其中同步每个 observable 的顺序。

public static IObservable<T> SynchronizeOrder<T>(IObservable<T> source, int order)
{
  //return synchronized source
}

更新:我最初的问题中有一点不清楚，即观察对象可以是不同的事件类型(这就是为什么我没有在我的示例中使用特定事件数据类型的原因)和因此我不想合并有序的可观察对象。我想要一种机制，可以确保独立的可观察对象按顺序同步。

myEventStream
 .ObserveOn(TaskFactory)
 .DoSomeExpensiveComputation()
 .Order(2) //doesn't have to be an extension method
 .Subscibe(computationResult=>
            sharedRessourceWhereOrderMatters.Update(computationResult))

Answer 1

这会使用可观察的连接执行您想要的操作:

    using System.Reactive;
    using System.Reactive.Subjects;
    using System.Reactive.Linq;

    var a = new Subject<int>();
    var b = new Subject<int>();
    var c = new Subject<int>();

    var test = 
        Observable.When(
            a.And(b).And(c).Then((a1, b1, c1) => new int[] { a1, b1, c1 })
        ).SelectMany(arr => arr.ToObservable());

    var sub = test.Subscribe(val => Console.Write("{0} ", val));

    a.OnNext(1);
    b.OnNext(2);
    c.OnNext(3);

    a.OnNext(1);
    c.OnNext(3);
    b.OnNext(2);

    c.OnNext(3);
    a.OnNext(1);
    b.OnNext(2);

    Console.ReadKey();
    sub.Dispose();

输出是:

1 2 3 1 2 3 1 2 3

但是请注意:

1. 这不提供允许您以某种顺序注入(inject)特定可观察对象的扩展方法。您需要使用 And(可以零碎地完成)来构建计划，然后最终执行它。 “建立”顺序将是输出顺序。

2. 这不会查看实际的窗口创建。但是，没有理由不能将它注入(inject)到可观察链的某处。

3. 每当所有三个可观察值都有值时，此解决方案将触发一次。您可能需要对其进行设置并在正确的位置将其拆除以获得所需的行为。