java - 使用重放(selectorFoo)但不使用发布(selectorFoo)时出现OOM

Question

这会因 OOM 崩溃:

Flowable.range(1, 5000)
        .map(__ -> new byte[1024 * 1024])
        .replay(
          fb ->
            fb.take(1)
              .concatMap(__ -> fb)
          ,1
        )
        .count()
        .toFlowable()
        .blockingSubscribe(c -> System.out.println("args = [" + c + "]"));

我认为这是因为 replay 保留了来自上游的排放，尽管我本以为 1 缓冲区大小提示不会这样做。 ...我错过了什么？

这不会崩溃:

Flowable.range(1, 5000)
        .map(__ -> new byte[1024 * 1024])
        .publish(
          fb ->
            fb.take(1)
              .concatMap(first -> fb.startWith(first))
          ,1
        )
        .count()
        .toFlowable()
        .blockingSubscribe(c -> System.out.println("args = [" + c + "]"));

但我不确定是否能保证我会得到上游的所有排放......

Answer 1

我对此进行了调查并找到了问题的原因:RxJava 2 中的 replay 中存在错误。

发生的情况是，replay 保存了对 2 个订阅者的引用，一个用于 take，另一个用于 concatMap 中的内部消费者。局部变量，因此从主线程到失效的 take 仍然引用第一个项目，有一个 GC 根。由于有界重播使用链接列表，因此第一个项目会通过其“下一个”链接不断引用越来越新的项目，最终耗尽内存。

publish 不会保留对旧值的引用，因此这不是问题。