我正在尝试对在不同线程上无序到达的事件进行重新排序。

是否可以创建与这些弹珠图相匹配的响应式扩展查询:

s1          1   2       3   4

s2          1   3   2       4

result      1       2   3   4

和...

s1          1   2   3   4

s2          4   3   2   1

result                  1234

即:只按版本号顺序发布结果。

我得到的最接近的是每次 s1 滴答时使用 Join 打开一个窗口，并且仅当 s2 以相同的数字到达时才关闭它。

像这样:

var publishedEvents = events.Publish().RefCount();
publishedEvents.Join(
        publishedEvents.Scan(0, (i, o) => i + 1),
        expectedVersion => publishedEvents.Any(@event => @event.Version == expectedVersion),
        _ => Observable.Never<Unit>(),
        (@event, expectedVersion) => new {@event,expectedVersion})
    .Where(x => x.expectedVersion == x.@event.Version)
    .Select(x => x.@event)
    .Subscribe(Persist);

但这不适用于图表 2。一旦 s2 与数字 2 同步，即在 1 之前，第 2 组将完成。

有意义吗？可以用 Rx 完成吗？应该吗？

编辑:我猜这就像重叠的窗口，在前面的所有窗口关闭之前，后面的窗口无法关闭。在窗口号与事件版本号匹配之前，前面的窗口不会关闭。

编辑 2:

我现在有类似的东西，但它并不是真正的响应式(Reactive)、函数式、线程安全的 LINQ 启示，我希望如此(请忽略我的事件现在是 JObjects):

var orderedEvents = Observable.Create<JObject>(observer =>
{
    var nextVersionExpected = 1;
    var previousEvents = new List<JObject>();
    return events
        .ObserveOn(Scheduler.CurrentThread)
        .Subscribe(@event =>
        {
            previousEvents.Add(@event);

            var version = (long) @event["Version"];
            if (version != nextVersionExpected) return;

            foreach (var previousEvent in previousEvents.OrderBy(x => (long) x["Version"]).ToList())
            {
                if ((long) previousEvent["Version"] != nextVersionExpected)
                    break;

                observer.OnNext(previousEvent);
                previousEvents.Remove(previousEvent);
                nextVersionExpected++;
            }
        });
});

最佳答案

简介

这个问题的关键是排序。不管怎么看，都需要某种形式的缓冲。虽然毫无疑问，一些精心设计的运算符组合可能会实现这一点，但我认为这是一个很好的例子，其中 Observable.Create是一个不错的选择。

概括解决方案

我已经做了一些努力来推广我的方法来接受任何类型的排序键。为此，我希望得到:

• 用于获取事件键的键选择器函数，类型为Func<TSource,TKey>
• TKey 类型的初始键
• 按顺序获取下一个键的函数，类型为 Func<TKey,TKey>
• 一个结果选择器，用于从源流中的配对事件中生成结果，类型为 Func<TSource,TSource,TSource>

由于我只是使用基于 1 的整数序列进行测试，因此满足以下条件:

• keySelector:i => i
• 第一个键:1
• nextKeyFunc: k => k+1
• 结果选择器:(left,right) => left

排序

这是我的 Sort试图。它将事件缓冲到字典中并尽快将它们刷新给订阅者:

public static IObservable<TSource> Sort<TSource, TKey>
    (this IObservable<TSource> source,
     Func<TSource, TKey> keySelector,
     TKey firstKey,
     Func<TKey, TKey> nextKeyFunc)
{
    return Observable.Create<TSource>(o =>
    {
        var nextKey = firstKey;
        var buffer = new Dictionary<TKey, TSource>();
        return source.Subscribe(i =>
        {
            if (keySelector(i).Equals(nextKey))
            {
                nextKey = nextKeyFunc(nextKey);
                o.OnNext(i);
                TSource nextValue;
                while (buffer.TryGetValue(nextKey, out nextValue))
                {
                    buffer.Remove(nextKey);
                    o.OnNext(nextValue);
                    nextKey = nextKeyFunc(nextKey);
                }
            }
            else buffer.Add(keySelector(i), i);
        });
    });
}

我不得不说这是一个非常幼稚的实现。在过去的生产代码中，我通过特定的错误处理、固定大小的缓冲区和超时来详细说明这一点，以防止资源泄漏。但是，它适用于此示例。 :)

有了这个排序(抱歉!)，我们现在可以考虑处理多个流。

合并结果

第一次尝试

我在这方面的第一次尝试是产生一个无序的事件流，这些事件已经被看到了所需的次数。然后可以对此进行排序。为此，我使用 GroupByUntil 按键对元素进行分组。保持每个组，直到两个元素被捕获。然后，每个组都是同一键的结果流。对于整数事件的简单示例，我可以只取每个组的最后一个元素。但是，我不喜欢这样，因为对于更真实的场景来说，每个结果流都可能贡献一些有用的东西，这很尴尬。为了感兴趣，我包含了代码。请注意，为了在这次尝试和我的第二次尝试之间共享测试，我接受了一个未使用的 resultSelector 参数:

public static IObservable<TSource> OrderedCollect<TSource, TKey>
    (this IObservable<TSource> left,
     IObservable<TSource> right,
     Func<TSource, TKey> keySelector,
     TKey firstKey,
     Func<TKey, TKey> nextKeyFunc
     Func<TSource,TSource,TSource> resultSelector)
{
    return left.Merge(right)
               .GroupByUntil(keySelector, x => x.Take(2).LastAsync())
               .SelectMany(x => x.LastAsync())
               .Sort(keySelector, firstKey, nextKeyFunc);
}

旁白:你可以破解SelectMany子句来决定如何选择结果。与第二次尝试相比，此解决方案的一个优势是，在具有许多结果流的情况下，更容易看到如何扩展它以选择三个结果元组中的前两个到达。

第二次尝试

对于这种方法，我独立地对每个流进行排序，然后 Zip结果在一起。这不仅是一个看起来简单得多的操作，而且以​​有趣的方式组合来自每个流的结果也容易得多。为了使测试与我的第一种方法兼容，我选择了 resultSelector 函数来使用第一个流的事件作为结果，但显然你可以灵活地做一些对你的场景有用的事情:

public static IObservable<TSource> OrderedCollect<TSource, TKey>
    (this IObservable<TSource> left,
     IObservable<TSource> right,
     Func<TSource, TKey> keySelector,
     TKey firstKey,
     Func<TKey, TKey> nextKeyFunc,
     Func<TSource, TSource, TSource> resultSelector)
{
    return Observable.Zip(
        left.Sort(keySelector, firstKey, nextKeyFunc),
        right.Sort(keySelector, firstKey, nextKeyFunc),
        resultSelector);
}

旁白:不难看出如何将此代码扩展到接受任意数量输入流的更一般情况，但如前所述，使用 Zip使得在所有流的结果都进入之前，在给定键处阻塞是非常不灵活的。

测试用例

最后，这是我的测试，与您的示例场景相呼应。要运行这些，请导入 nuget 包 rx-testing和 nunit并将上面的实现放入静态类中:

public class ReorderingEventsTests : ReactiveTest
{
    [Test]
    public void ReorderingTest1()
    {
        var scheduler = new TestScheduler();

        var s1 = scheduler.CreateColdObservable(
            OnNext(100, 1),
            OnNext(200, 2),
            OnNext(400, 3),
            OnNext(500, 4));

        var s2 = scheduler.CreateColdObservable(
            OnNext(100, 1),
            OnNext(200, 3),
            OnNext(300, 2),
            OnNext(500, 4));

        var results = scheduler.CreateObserver<int>();

        s1.OrderedCollect(
            right: s2,
            keySelector: i => i,
            firstKey: 1,
            nextKeyFunc: i => i + 1,
            resultSelector: (left,right) => left).Subscribe(results);

        scheduler.Start();

        results.Messages.AssertEqual(
            OnNext(100, 1),
            OnNext(300, 2),
            OnNext(400, 3),
            OnNext(500, 4));
    }

    [Test]
    public void ReorderingTest2()
    {
        var scheduler = new TestScheduler();

        var s1 = scheduler.CreateColdObservable(
            OnNext(100, 1),
            OnNext(200, 2),
            OnNext(300, 3),
            OnNext(400, 4));

        var s2 = scheduler.CreateColdObservable(
            OnNext(100, 4),
            OnNext(200, 3),
            OnNext(300, 2),
            OnNext(400, 1));

        var results = scheduler.CreateObserver<int>();

        s1.OrderedCollect(
            right: s2,
            keySelector: i => i,
            firstKey: 1,
            nextKeyFunc: i => i + 1,
            resultSelector: (left, right) => left).Subscribe(results);

        scheduler.Start();

        results.Messages.AssertEqual(
            OnNext(400, 1),
            OnNext(400, 2),
            OnNext(400, 3),
            OnNext(400, 4));
    }
}

柯里化(Currying)以避免重复

最后的评论，因为我讨厌在代码中重复自己，这里有一个调整可以避免我调用 Sort 的重复方式在第二种方法中。我没有将它包含在正文中，以免让不熟悉柯里化(Currying)的读者感到困惑:

public static IObservable<TSource> OrderedCollect<TSource, TKey>
    (this IObservable<TSource> left,
        IObservable<TSource> right,
        Func<TSource, TKey> keySelector,
        TKey firstKey,
        Func<TKey, TKey> nextKeyFunc,
        Func<TSource, TSource, TSource> resultSelector)
{
    Func<IObservable<TSource>, IObservable<TSource>> curriedSort =
        events => events.Sort(keySelector, firstKey, nextKeyFunc);

    return Observable.Zip(
        curriedSort(left),
        curriedSort(right),
        resultSelector);
}

关于c# - 使用 Reactive Extensions 重新排序事件，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/19939258/