java - RxJava:merge()更改发射项目的顺序？

看一下这两个小测试:

@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，结果是相同的。

我想念什么？

谢谢。

最佳答案

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

从flatMap doc:

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

从merge doc:

合并可能会交织合并的Observable发出的项目(类似的运算符Concat不会交织项目，而是依次开始发出每个源Observable的所有项目，然后才开始从下一个来源Observable发出项目。)

引用此SO Answer:

在您的情况下，使用单元素静态流不会产生任何实际的变化(但从理论上讲，合并可能会以随机顺序输出单词，但根据规范仍然有效)

如果您需要有保证的订单，请使用concat*。

第一个例子

它是这样的:

发出1时，groupBy运算符将使用键GroupedObservable创建一个false

flatMap将输出此可观察项中的项目-当前仅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中使用

，您可以使用地图创建一个新的可观察对象，在这里您首先添加偶数元素(key = true)，然后添加奇数元素(key = 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