RxJava - interval操作符

Observable.interval(0, 1, TimeUnit.SECONDS)

参数一：什么时候开始发送 参数二：隔多长时间发一次 参数三：计时的单位，是秒还是毫秒

/** * 定时执行 * * @param initialDelay 第一次执行的延迟时间 * @param period 每次执行的间隔的时间 * @param unit 时间单位 * @return Observable对象 */ public static Observable<Long> interval(long initialDelay, long period, TimeUnit unit) take(count)

控制观察者接收事件的数量。take就是发射数据后控制什么时候停下来

map

map可以将被观察者发送的数据转变成其他的类型，比如Long 转成 String类型

doOnSubscribe

在 subscribe() 调用后而且在事件发送前执行；默认情况下， doOnSubscribe() 执行在 subscribe() 发生的线程,Observable每发送onSubscribe( )之前都会回调这个

 

/** * 发送验证码倒计时 * * @param view */ final int count = 60; Observable.interval(0, 1, TimeUnit.SECONDS) .take(count) .map(new Function<Long, String>() { @Override public String apply(Long aLong) throws Exception { Log.e(TAG, "apply: "+aLong ); return String.valueOf(count - aLong); } }) .observeOn(AndroidSchedulers.mainThread()) .doOnSubscribe(new Consumer<Disposable>() { @Override public void accept(Disposable disposable) throws Exception { Log.e(TAG, "accept: "); } }) .subscribe(new Observer<String>() { @Override public void onSubscribe(Disposable d) { } @Override public void onNext(String string) { Log.e(TAG, "onNext: "+string ); } @Override public void onError(Throwable e) { } @Override public void onComplete() { } });

日志打印结果：

07-02 17:00:05.122 19762-19762**** E/MainActivity: accept: 07-02 17:00:05.124 19762-19783**** E/MainActivity: apply: 0 07-02 17:00:05.152 19762-19762**** E/MainActivity: onNext: 60 07-02 17:00:06.123 19762-19783**** E/MainActivity: apply: 1 07-02 17:00:06.123 19762-19762**** E/MainActivity: onNext: 59 07-02 17:00:07.123 19762-19783**** E/MainActivity: apply: 2 07-02 17:00:07.124 19762-19762**** E/MainActivity: onNext: 58 ************************************************ 07-02 17:01:03.123 19762-19783**** E/MainActivity: apply: 58 07-02 17:01:03.124 19762-19762**** E/MainActivity: onNext: 2 07-02 17:01:04.123 19762-19783**** E/MainActivity: apply: 59 07-02 17:01:04.123 19762-19762**** E/MainActivity: onNext: 1

还有一种就是每隔一段时间就会发送一个事件，这个事件是从0开始，不断增1的数字

/** * 定时执行 * * @param period 每次执行的间隔的时间 * @param unit 时间单位 * @return Observable对象 */ public static Observable<Long> interval(long period, TimeUnit unit)

例子：

Disposable disposable; Observable.interval(1, TimeUnit.SECONDS).subscribe(new Observer<Long>() { @Override public void onSubscribe(Disposable d) { disposable = d; } @Override public void onNext(Long aLong) { Log.e(TAG, "onNext: " + aLong); if (aLong == 10) { disposable.dispose();//等于10的取消事件的发送 } } @Override public void onError(Throwable e) { } @Override public void onComplete() { } });

日志打印结果：

07-02 17:17:57.300 20799-20820-------------- E/MainActivity: onNext: 0 07-02 17:17:58.300 20799-20820-------------- E/MainActivity: onNext: 1 07-02 17:17:59.300 20799-20820-------------- E/MainActivity: onNext: 2 07-02 17:18:00.300 20799-20820-------------- E/MainActivity: onNext: 3 07-02 17:18:01.300 20799-20820-------------- E/MainActivity: onNext: 4 07-02 17:18:02.300 20799-20820-------------- E/MainActivity: onNext: 5 07-02 17:18:03.300 20799-20820-------------- E/MainActivity: onNext: 6 07-02 17:18:04.300 20799-20820-------------- E/MainActivity: onNext: 7 07-02 17:18:05.300 20799-20820-------------- E/MainActivity: onNext: 8 07-02 17:18:06.300 20799-20820-------------- E/MainActivity: onNext: 9 07-02 17:18:07.300 20799-20820-------------- E/MainActivity: onNext: 10