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

Question

根据 groupBy 的文档:

Note: A GroupedObservable will cache the items it is to emit until such time as it is subscribed to. For this reason, in order to avoid memory leaks, you should not simply ignore those GroupedObservables that do not concern you. Instead, you can signal to them that they may discard their buffers by applying an operator like take(int)(0) to them.

有一个 RxJava tutorial其中说:

Internally, every Rx operator does 3 things

1. It subscribes to the source and observes the values.
2. It transforms the observed sequence according to the operator's purpose.
3. It pushes the modified sequence to its own subscribers, by calling onNext, onError and 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 并且不会一直重新创建组本身。