RxJava：merge() 改变了发射项目的顺序？

Question

看看这两个小测试：

@Test
public void test1() {
    Observable.range(1, 10)
        .groupBy(v -> v % 2 == 0)
        .flatMap(group -> {
            if (group.getKey()) {
                return group;
            }
            return group;
        })
        .subscribe(System.out::println);
}

@Test
public void test2() {
    Observable.range(1, 10)
        .groupBy(v -> v % 2 == 0)
        .toMap(g -> g.getKey())
        .flatMapObservable(m -> Observable.merge(
            m.get(true),
            m.get(false)))
        .subscribe(System.out::println);
}

我希望两者都以相同的顺序返回一个数字列表，所以：

1 2 3 4 5 6 7 8 9 10

但第二个例子返回

2 4 6 8 10 1 3 5 7 9

改为。

似乎在第二个示例中merge 正在执行concat，实际上如果我将其更改为concat，结果是相同的。

我错过了什么？

谢谢。

Answer 1

基本上flatMap和merge不保证发射项目的顺序。

来自flatMap文档：

请注意，FlatMap 会合并这些 Observable 的发射，以便它们可以交错。

来自merge文档：

Merge 可以交错由合并后的 Observable 发出的项（类似的运算符 Concat 不会交错项，而是在开始从下一个源 Observable 发出项之前依次发出每个源 Observable 的所有项）。

引用此SO Answer：

在您的情况下，对于单元素静态流，它没有任何真正的区别（但理论上，合并可以以随机顺序输出单词并且仍然根据规范有效）

如果您需要保证订单，请改用concat*。

第一个例子

它是这样工作的：

• 当发出1 时，groupBy 运算符将创建一个GroupedObservable，其键为false
  • flatMap 将从这个 observable 输出项目 - 目前只有 1
• 当发出2 时，groupBy 运算符将创建一个GroupedObservable，其键为true
  • flatMap 现在也将输出第二个 GroupedObservable 的项目 - 目前是 2
• 当发出3 时，groupBy 运算符会将其添加到现有的GroupedObservable 中，键为false 和flatMap 将立即输出此项目
• 当发出4 时，groupBy 运算符会将其添加到现有的 GroupedObservable 中，键为 true 和 flatMap 将立即输出此项目
• 等

它可能会帮助您添加更多日志记录：

    Observable.range(1, 10)
            .groupBy(v -> v % 2 == 0)
            .doOnNext(group -> System.out.println("key: " + group.getKey()))
            .flatMap(group -> {
                if (group.getKey()) {
                    return group;
                }
                return group;
            })
            .subscribe(System.out::println);

那么输出是：

key: false
1
key: true
2
3
...

第二个例子

这是完全不同的，因为toMap 将阻塞直到上游完成：

• 当发出1 时，groupBy 运算符将创建一个GroupedObservable，其键为false
  • toMap 将把这个GroupedObservable 添加到内部映射并使用键false（与GroupedObservable 具有相同的键）
• 当发出2 时，groupBy 运算符将创建一个GroupedObservable，其键为true
  • toMap 将把这个GroupedObservable 添加到内部映射并使用键true（与GroupedObservable 具有相同的键） - 所以现在映射有2 个GroupedObservables
• 以下数字被添加到相应的GroupedObservables 中，当源代码完成时，toMap 运算符完成并将映射传递给下一个运算符
• 在flatMapObservable 中，您使用映射创建一个新的可观察对象，您首先添加偶数元素（键 = true），然后添加奇数元素（键 = false）

您还可以在此处添加更多日志记录：

    Observable.range(1, 10)
            .groupBy(v -> v % 2 == 0)
            .doOnNext(group -> System.out.println("key: " + group.getKey()))
            .toMap(g -> g.getKey())
            .doOnSuccess(map -> System.out.println("map: " + map.size()))
            .flatMapObservable(m -> Observable.merge(
                    m.get(true),
                    m.get(false)
            ))
            .subscribe(System.out::println);

那么输出是：

key: false
key: true
map: 2
2
4
6
8
10
1
3
5
7
9