RXJava - hot observables - 安全的 CPU 资源

Question

这是一个简单的可观察示例：

observable
            .filter(...)
            .buffer(50, TimeUnit.MILLISECONDS)
            .doOnNext(/* this is executed all the time... */)
            .filter(data -> data.size() > 0)
            .doOnNext(/* this is not executed because of the filter above... */)
            .flatMap(data -> Observable.from(data).distinctUntilChanged())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe();

问题/疑问

我从 buffer(...) 函数得到的 observable 是一个又一个结果，大部分是空的。

用这样的 observables 过滤是处理这个问题的正确方法吗？同时拥有大量此类订阅是否会降低成本？还是应该以不同的方式处理？

Answer 1

一般来说，是的，过滤掉下游不需要的东西。

同时拥有大量此类订阅是否会降低成本？

技术上：是的。更多的计时器，计算线程上更多的活动序列。但是，序列的开销可能远小于您执行的业务逻辑。

为了节省一些开销，我会使用flatMapIterable：

observable
    .filter(...)
    .buffer(50, TimeUnit.MILLISECONDS)
    .doOnNext(/* this is executed all the time... */)
    .filter(data -> data.size() > 0)
    .doOnNext(/* this is not executed because of the filter above... */)
    .flatMapIterable(data -> data)               // <-----------------------------
    .distinctUntilChanged()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe();