同步多个可观察量的顺序

本文关键字：顺序 观察 同步 | 更新日期: 2023-09-27 18:31:40

我有多个（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等。

窗口

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

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

扩展问题（不那么重要）：可观察量彼此独立，因此我需要引入一种通用的静态方法，我可以在其中同步每个可观察量的顺序。

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))

同步多个可观察量的顺序

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

    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. 这不提供允许您按某种顺序注入特定可观察量的扩展方法。您需要使用And构建计划（可以逐段完成），然后最终执行它。"建立"顺序将是输出顺序。

  2. 这不会查看实际的窗口创建。但是，没有理由不能注入到某个可观察链的某个地方。

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

Yamens 的答案非常适合同一范围内具有相同类型的可观察量。在我的情况下（扩展问题），可观察量可以是不同的类型，并且可以在执行的任何点同步顺序。

所以这是我提出的简单解决方案：

private readonly object syncRoot = new object();
private int sourceCount;
private readonly SortedDictionary<int, Action> buffer = new SortedDictionary<int, Action>();
public IObservable<T> Order<T>(IObservable<T> source, int order)
{
    return Observable.Create<T>(observer =>
    {
        lock (syncRoot)
            sourceCount++;
        source.Synchronize(syncRoot).Subscribe(item =>
        {
            buffer[order] = () => observer.OnNext(item);
            if(buffer.Count == sourceCount)
            {
                foreach (var action in buffer.Values)
                    action();
                buffer.Clear();
            }
        }, observer.OnError, observer.OnCompleted);
        return () =>
        {
            lock (syncRoot)
                sourceCount--;
        };
    });
}//end order
相关文章：