深入RxJava之时间管理大师

[soapgu]

• 序言

很久没更新关于ReactiveX系列的技术Blog了，正好最近又用到一些既好用又略“冷门”api，所以正好“组团”揍一波更新下，正好做成一篇“时间管理大师”，感慨下RxJava对时间坐相关精确控制的能力。我决定从常用到冷门列举。

• 初级时间管理之延时投放

• Timer

create an Observable that emits a particular item after a given delay

• 代码用例

Observable<Long> eggTimer = Observable.timer(5, TimeUnit.MINUTES);

eggTimer.blockingSubscribe(v -> System.out.println("Egg is ready!"));

• 使用场景
  就是延时触发，比较常见。
  比如有时候数据到了，还需要延时一段时间等UI准备好。
  比如开发一个窗体后，需要延时一段时间自动关闭等。

• 链接

• timer

• 初级时间管理之定时投放

• Interval

create an Observable that emits a sequence of integers spaced by a given time interval

• Code Sample

Observable<Long> clock = Observable.interval(1, TimeUnit.SECONDS);

clock.subscribe(time -> {
    if (time % 2 == 0) {
        System.out.println("Tick");
    } else {
        System.out.println("Tock");
    }
});

• 使用场景
  比较常见是替换Timer控件使用
  定时刷新
  定时任务等

• 链接

• interval

• 中级时间管理之——过时不侯

• 曾经端小烦恼

我有过一次这样的场景，王需要某一个http的请求需要设置timeout时间为1秒，而其他的请求设置为标准1分钟。
OKhttp确实可以设置超时时间，但是为了提高性能用的都是共享实例，你让我单独为一个路由设置超时时间，臣妾做不到啊～～～。

• Timeout api
  
  The Timeout operator allows you to abort an Observable with an onError termination if that Observable fails to emit any items during a specified span of time.

• 解析
  timeout不仅可以用来控制那种http请求response这样的“一次性”用途，还可以控制Observable这种连续的Rx，只要数据推送“慢”了，就能截断。有个这个api我就可以，直接用来设置单个路由的超时，还不用“关心”http 组件的api。用更抽象的方法去解决问题，简直就是“降维打击”。

另外这个api正好和timer反着来，自己意会一下

• Code Sample

    public Single<Boolean> checkOnline() {
        Logger.i("---being check online---");
        return this.login.checkOnline()
                .timeout(1, TimeUnit.SECONDS)
                .andThen(Single.just(true))
                .onErrorReturnItem(false)
                .flatMap( t ->{
                    online = t;
                    Logger.i("---end check online,result:%s---",t);
                    return Single.just(online);
                } );
    }

注：这里使用了onErrorReturnItem，把java.util.concurrent.TimeoutException 处理了用false代替代表不在线。

• 链接

• timeout

• 高级时间管理之——限流器

在实际开发过程中，“限流”的算法其实是蛮难表达的。这里有多种限流算法，总有一款适合您

配图无关～～～

• 低配版——定时采样

可以理解为在定时时间内，只采集一个值，把其他的值全部过滤掉

• api

• throttleFirst
  就是定期只采定时内第一个值

• throttleLast
  就是定期只采定时内最后一个值

• throttleLatest
  和throttleLast基本等价，区别就是会第一时间把第一个值放出来

• 高配版——超时采样

// Diagram:
// -A--------------B----C-D-------------------E-|---->
//  a---------1s
//                 b---------1s
//                      c---------1s
//                        d---------1s
//                                            e-|---->
// -----------A---------------------D-----------E-|-->

Observable<String> source = Observable.create(emitter -> {
    emitter.onNext("A");

    Thread.sleep(1_500);
    emitter.onNext("B");

    Thread.sleep(500);
    emitter.onNext("C");

    Thread.sleep(250);
    emitter.onNext("D");

    Thread.sleep(2_000);
    emitter.onNext("E");
    emitter.onComplete();
});

source.subscribeOn(Schedulers.io())
        .throttleWithTimeout(1, TimeUnit.SECONDS)
        .blockingSubscribe(
                item -> System.out.println("onNext: " + item),
                Throwable::printStackTrace,
                () -> System.out.println("onComplete"));

// prints:
// onNext: A
// onNext: D
// onNext: E
// onComplete

看图更容易理解，可以过滤段时间内的大量抖动数据，只有当数据发送不那么频繁，才开始“吐数据”

• 适合场景
  过滤连续按按钮等场景

• api

• debounce

• throttlewithtimeout

注：上面两个api完全等价。建议使用debounce，和苹果的Combine的命名一致。

• 总结体会

ReactiveX的框架确实是一个非常好的异步管理框架。越用越香，现在的收益越来越明显。

1. 前期困难，后面学习难度非常平缓
2. 抽象度很高，可以解决很多代码难写的场景
3. 基本上各个平台都有应用，应用范围非常广泛