如何使用 Reactor 处理历史数据？

Question

假设我有两个历史事件来源，并且每个来源的事件都按时间顺序排列。

如何使用 Reactor 合并这些源，以便合并 Flux 中的事件按时间顺序发出？

在 RxNET 中，Observable.Generate() 和 HistoricalScheduler 的组合可用于从历史事件源创建 Observable，以便根据事件的时间安排排放（详细信息 here） ，但我无法在 Reactor 中找到等效的方法。

也许我可以以某种方式将Flux.generate() 与VirtualTimeScheduler 一起使用？

下面是一个玩具示例：

public class Program {

  public record Event(Instant time, String id) {}

  public static void main(String[] args) {

    var source1 = Arrays.asList(
        new Event(Instant.ofEpochMilli(10), "a"),
        new Event(Instant.ofEpochMilli(30), "c"),
        new Event(Instant.ofEpochMilli(50), "e")
    );

    var source2 = Arrays.asList(
        new Event(Instant.ofEpochMilli(20), "b"),
        new Event(Instant.ofEpochMilli(40), "d"),
        new Event(Instant.ofEpochMilli(60), "f")
    );

    Flux.fromIterable(source1)
        .mergeWith(Flux.fromIterable(source2))
        .subscribe(e -> System.out.println(e.id));

    // current output:
    // a
    // c
    // e
    // b
    // d
    // f

    // desired output:
    // a
    // b
    // c
    // d
    // e
    // f
   
  }
}

Answer 1

您可以使用mergeComparingWith 运算符并像这样提供Comparator：

Flux.fromIterable(source1)
         .mergeComparingWith(Flux.fromIterable(source2), Comparator.comparing(Event::time, Instant::compareTo))
         .subscribe(e -> System.out.println(e.id));

它通过从每个序列中选取最小值来生成重新排序的合并序列。