RXJava 無限ストリームでdistinct使ったらどうなる？

検証バージョン

• Java 1.8.0_25
• RxJava 2.1.14

distinct

distinct はフィルタ系操作の一つで、重複要素を取り除くためのメソッドです。

重複要素を取り除くためにはどこかに比較対象が存在しないといけないわけで、じゃあ無限ストリームとかどうやって対応するんだろうか、と思いませんか。皆さん。

メモリリークします。以上です。

public class RxJava {
  @Test
  public void test() throws InterruptedException {
    Flowable.interval(1, TimeUnit.MILLISECONDS)
        .distinct()
        .subscribe(new DebugSubscriber());
    TimeUnit.SECONDS.sleep(100000);
  }
}

Javadocにも書いてますね。

Note that this internal HashSet may grow unbounded as items won’t be removed from it by the operator. Therefore, using very long or infinite upstream (with very distinct elements) may lead to OutOfMemoryError.

distinctUntilChanged

じゃあメモリリークしないためには比較対象を絞ればいいよね、というのがdistinctUntilChanged。

これは直前のデータしか保持しないため、OOMになる心配がない。

distinctのカスタマイズ

distinct には独自のCollectionを渡すことができる。そのため、もう少し比較対象を拡大したければリミットを設けた自作Collectionを作ってもいい。が、自作Collectionでバグを埋め込まないように注意。