a0973378273的部落格

RxJava2的observable需要取消訂閱，否則會有memory leak的風險.

.subscribeOn變換上面function的執行序(只能改一次)(往下遞延)

gradle新增依賴

implementation "io.reactivex.rxjava2:rxjava:2.1.7"
implementation "io.reactivex.rxjava2:rxandroid:2.0.1"

每次delay1秒重複發射RX流，直到return true;

Observable.repeatUntil(() -> {Thread.sleep(1000);return false;})

return timer會一直重複發射RX流，直到發射onComplete時候停止

.repeatWhen(objectObservable ->
    objectObservable.flatMap(o -> {
        if(++index > 5) {
            return Observer::onComplete;
        }else{
            return objectObservable.timer(1,TimeUnit.SECONDS);
        }
    })
)

用buffer搜集，再用fromIterable+reduce去阻塞每個Rx流做處理，之後用concatMap(flatMap)發射一次

Observable.range(1, 100)
.buffer(50)
.concatMap(int-> Observable.fromIterable(int).reduce((s, s2) -> s + s2).toObservable())