Rx 中的 groupBy、过滤器和内存泄漏

Question

根据groupBy的文档：

注意：GroupedObservable 将缓存它要发出的项目，直到它被订阅。因此，为了避免内存泄漏，您不应该简单地忽略那些与您无关的GroupedObservables。相反，您可以向他们发出信号，告知他们可能会通过对他们应用take(int)(0) 之类的运算符来丢弃缓冲区。

有一个RxJava tutorial 说：

在内部，每个 Rx 操作员做 3 件事

1. 它订阅源并观察值。
2. 它根据操作员的目的转换观察到的序列。
3. 它通过调用 onNext、onError 和 onCompleted 将修改后的序列推送给自己的订阅者。

让我们看看下面的代码块，它只从range(0, 10)中提取偶数：

Observable.range(0, 10)
        .groupBy(i -> i % 2)
        .filter(g -> g.getKey() % 2 == 0)
        .flatMap(g -> g)
        .subscribe(System.out::println, Throwable::printStackTrace);

我的问题是：

1. 这是否意味着filter 运营商已经暗示订阅来自groupBy 的每个组或只是Observable<GroupedObservable> 一个？

2. 在这种情况下会不会出现内存泄漏？如果是这样，

3. 如何正确丢弃这些组？将filter 替换为自定义的take(0)，后跟return Observable.empty()？你可能会问我为什么不直接返回take(0)：这是因为filter 不一定紧跟在groupBy 之后，而是可以在链中的任何位置并涉及更复杂的条件。

Answer 1

除了内存泄漏之外，由于内部请求协调问题，当前的实现可能最终完全挂起。

请注意，使用take(0)，可能会一直重新创建组。我会改用 ignoreElements 来丢弃值，没有项目到达 flatMap 并且不会一直重新创建组本身。

Answer 2

您的怀疑是正确的，因为要正确处理分组的可观察对象，必须订阅每个内部可观察对象 (g)。由于filter 只订阅外部可观察对象，所以这是个坏主意。只需在flatMap 中执行您需要的操作，使用ignoreElements 过滤掉不需要的组。

Observable.range(0, 10)
    .groupBy(i -> i % 2)
    .flatMap(g -> {
       if (g.getKey() % 2 == 0) 
         return g;
       else 
         return g.ignoreElements();
    })
    .subscribe(System.out::println, Throwable::printStackTrace);