如何创建两个排序后的可观察对象之间差异的可观察对象？答案

【问题标题】：How to create an observable of the difference between two sorted observables?如何创建两个排序后的可观察对象之间差异的可观察对象？
【发布时间】：2017-05-17 15:56:41
【问题描述】：

假设我有两个可观察的排序双精度数。我想将它们之间的区别作为可观察的。例如：

           1       2           4        
 left:   ──o───────o───────────o────/
           1    3  4   5
 right:  ──o────o──o───o───/
                   l2          r3   r5
 output: ──────────o───────────o────o─/

对此的命令式实现很简单：在您尚未到达的一侧保留一个项目列表，并从另一侧“发出”这些项目。

在 RFP 世界中，解决此问题的规范方法是什么？我专门使用 RxScala。

【问题讨论】：

您能否提供一个大理石图，显示来自源的值何时发出以及如何导致结果 observable？
@Enigmativity 我添加了一个大理石图。 HTH
那是一张漂亮的大理石图。你是怎么做到的？我仍然无法弄清楚规则。你能解释一下吗？
@Enigmativity 这是一个手工制作的大理石图，仅由最好的 ASCII 字符制成（也就是我手工制作的）。简单的解释是，我想要两个 observables 之间的区别，排序，并且只要我能理解它在任何一方都不存在。
您可能需要再试一次 - “我想要两个可观察对象之间的区别，排序并且一旦我能理解它在任何一方都不存在” - 那只是不对我来说一点意义都没有，抱歉。

标签： rx-java reactive-programming observable reactivex rx-scala

【解决方案1】：

这就是我在 rxjava 中的做法，这意味着两个 observable 具有相同的长度。

    Observable<Integer> obs1 = Observable.just(1, 2, 4, 6);
    Observable<Integer> obs2 = Observable.just(1, 3, 4, 5);

    obs1.zipWith(obs2, (integer, integer2) -> {
        if (!Objects.equals(integer, integer2)) {
            return Observable.just(integer).concatWith(Observable.just(integer2));
        } else {
            return Observable.empty();
        }
    })
      .flatMap(observable -> observable)
      .sorted()
      .forEach(System.out::println);

编辑

另一种方法是使用集合

    Observable<Integer> obs1 = Observable.just(1, 2, 4);
    Observable<Integer> obs2 = Observable.just(1, 3, 4, 5);


    obs1.mergeWith(obs2)
            .sorted()
            .reduce(new ArrayList<Integer>(), (integers, integer) -> {
                if (integers.contains(integer)) {
                    integers.remove(integer);
                } else {
                    integers.add(integer);
                }
                return integers;
            })
            .flatMapIterable(integers -> integers)
            .forEach(System.out::println);

【讨论】：

您在这里做了两个假设：1. 可观察对象的长度相同（我已经更新了问题以表明它们不是），并且； 2. 一个 observables 不能遥遥领先于另一个（它们以相同的速度前进）。
这种新方法在两个可观察对象完成之前不需要阻塞吗？
@OmervanKloeten 不，没有阻塞，所有操作员都在生成非阻塞的 observables。
我的意思是“排序”运算符在发出第一个结果之前等待所有元素从两个来源返回。
@OmervanKloeten 您可以删除它并获得相同的结果

【解决方案2】：

我认为 RxJava 中没有太多东西可以“实时”实现这种对称差异...主要是因为通用运算符无法做出您的初始假设（可观察值已排序），因此很少有人可以帮助您那个。

但是，您可以尝试为此目的编写一个自定义运算符，从 sequenceEqual 获得灵感：它在内部逐步推进两个发布者之间的相等比较，因此这与您想要做的很接近.

【讨论】：