java - Concat VS 合并运算符

Question

我检查了 RXJava 的文档，我注意到 concat 和 merge 运算符似乎做同样的事情。我写了几个测试来确定。

@Test
public void testContact() {

    Observable.concat(Observable.just("Hello"),
                      Observable.just("reactive"),
                      Observable.just("world"))
              .subscribe(System.out::println);
}

@Test
public void testMerge() {

    Observable.merge(Observable.just("Hello"),
                      Observable.just("reactive"),
                      Observable.just("world"))
            .subscribe(System.out::println);
}

文档说

The Merge operator is also similar. It combines the emissions of two or more Observables, but may interleave them, whereas Concat never interleaves the emissions from multiple Observables.

但我还是不完全明白，运行这个测试数千次合并结果总是一样的。由于未授予订单，因此我期望有时会出现“响应式(Reactive)”“世界”“你好”。

代码在这里https://github.com/politrons/reactive

Answer 1

正如您引用的文档中所述 - merge 可以交错输出，而 concat 将首先等待较早的流完成，然后再处理以后的流。在您的情况下，对于单元素静态流，它没有任何真正的区别(但理论上，合并可以以随机顺序输出单词并且根据规范仍然有效)。如果您想看到差异，请尝试以下操作(您需要在之后添加一些 sleep 以避免提前退出)

    Observable.merge(
            Observable.interval(1, TimeUnit.SECONDS).map(id -> "A" + id),
            Observable.interval(1, TimeUnit.SECONDS).map(id -> "B" + id))
    .subscribe(System.out::println);

A0 B0 A1 B1 B2 A2 B3 A3 B4 A4

对

    Observable.concat(
            Observable.interval(1, TimeUnit.SECONDS).map(id -> "A" + id),
            Observable.interval(1, TimeUnit.SECONDS).map(id -> "B" + id))
    .subscribe(System.out::println);

A0 A1 A2 A3 A4 A5 A6 A7 A8

Concat 永远不会开始打印 B，因为流 A 永远不会结束。

s/stream/observable/g ;)

文档提供了很好的图表来显示差异。您需要记住，merge 没有保证一个一个地交错项目，它只是一个可能的例子。

连接

合并