Android开发中必不可少会遇到轮询或定时任务，在RxJava诞生之前，我们常常使用Handler+postDelay，或者Java中的Timer来实现，实际上RxJava也可以实现这类需求。下面，我们将分别介绍这几种方案的实现原理。
方案一：使用Handler实现轮询。
Handler提供了postDelay方法可以延迟执行某个Runnable，如果我们在Runnable的run方法中继续将当前Runnable postDelay到Handler中，则可以实现轮询。
下面代码将会间隔1秒打印从1到10。
Handler handler = new Handler(Looper.getMainLooper()); // 全局变量
int count = 0; // 全局变量
void testCount() {
handler.postDelay(new Runnable() {
void run() {
// do something
count++;
Log.d(TAG, "count: " + count);
if (count < 10) {
handler.postDelay(this, 1000);
}
}
}, 1000);
}
原理如上，但是实际上开发过程，我们需要考虑更多，比如如何方便的取消任务，当Activity销毁时候，要及时取消任务以防止内存泄漏，或者出现其他异常导致崩溃。另外，上面代码对于每个任务都需要去手动调用postDelay，属于重复性工作。我们可以对其进行封装，抽象出轮询以及取消的接口，而不需要管内部的实现。
下面我封装一个HandlerTimer类，内部提供下面的接口：
// 延迟执行任务
public TimerTask schedule(final Runnable runnable, final long delay, final TimeUnit delayTimeUnit);
// 轮询任务
public TimerTask schedule(final Runnable runnable, final long delay, final long period, final TimeUnit timeUnit);
// 取消任务
public void cancel(TimerTask timerTask);
schedule方法一个支持延迟执行任务，一个支持轮询，两个方法都返回一个TimerTask类，这个类有个cancel方法可以取消任务执行。当然，HandlerTimer类也提供了cancel接口来取消任务。
源码如下：
public class HandlerTimer {
private Handler handler;
public HandlerTimer(Handler handler) {
this.handler = handler;
}
public TimerTask schedule(final Runnable runnable, final long delay, final TimeUnit delayTimeUnit) {
TimerTask timerTask = new TimerTask() {
@Override
public void doRun() {
runnable.run();
}
};
handler.postDelayed(timerTask, delayTimeUnit.toMillis(delay));
return timerTask;
}
public TimerTask schedule(final Runnable runnable, final long delay, final long period, final TimeUnit timeUnit) {
TimerTask timerTask = new TimerTask() {
@Override
public void doRun() {
runnable.run();
handler.postDelayed(this, timeUnit.toMillis(period));
}
};
handler.postDelayed(timerTask, timeUnit.toMillis(delay));
return timerTask;
}
public void cancel(TimerTask timerTask) {
if (timerTask != null) {
timerTask.cancel();
handler.removeCallbacks(timerTask);
}
}
public static abstract class TimerTask implements Runnable {
private volatile boolean isCancelled;
public abstract void doRun();
public void cancel() {
isCancelled = true;
}
@Override
public void run() {
if (!isCancelled) {
doRun();
}
}
}
}
我们需要传入Handler来构造HandlerTimer类，TimerTask运行在Handler的Looper中，所以如果你的任务比较耗时，切记不要传入MainLooper的Handler，以免产生ANR。可以构造一个HandlerThread，利用其内部的Looper来构造Handler，这样TimerTask则运行在HandlerThread的内部线程中。另外，要记住不用的时候（比如Activity销毁时）还需要调用handlerThread.quit方法来停止线程运行避免内存泄漏。原理如下：
HandlerThread handlerThread = new HandlerThread("loop-timer");
handlerThread.start();
Handler handler = new Handler(handlerThread.getLooper());
HandlerTimer handlerTimer = new HandlerTimer(handler);
...
OK，使用我们封装的类来实现从1打印到10。
Handler handler = new Handler(Looper.getMainLooper()); // 全局变量
int count = 0; // 全局变量
HandlerTimer handlerTimer = new HandlerTimer(handler);
TimeTask timeTask;
void testCount() {
timerTask = handlerTimer.schedule(new Runnable() {
@Override
public void run() {
count++;
Log.d(TAG, "count: " + count);
if (count == 10) {
handlerTimer.cancel(timerTask);
}
}
}, 1, 1, TimeUnit.SECONDS);
}
方案二：使用Java的Timer和TimerTask实现轮询。
代码写法同我们封装的HandlerTimer差不多。
Timer timer = new Timer();
TimerTask countDownTask;
void testCount() {
if (countDownTask != null) {
countDownTask.cancel();
}
count = 0;
countDownTask = new TimerTask() {
@Override
public void run() {
count++;
if (count == 10) {
countDownTask.cancel();
}
}
};
timer.schedule(countDownTask, 1000, 1000);
}
void stopTimer() {
if (timer != null) {
timer.cancel();
}
}
Java的Timer内部有且仅有一个线程，用于执行TimerTask，使用完成后记得调用Timer的cancel方法来关闭整个Timer。单个任务的执行时长，也会影响其他任务的执行。当然上面的HandlerTimer也是如此。因为任务运行在子线程中，如果有更新UI的需求，可以利用Handler post到主线程中执行。
方案三：利用RxJava实现轮询
RxJava的interval操作符可每隔一定时间发射数据，数据从0开始。所以从1打印到10，我们可以这么写：
Disposable disposable; // 全局变量
void testCount() {
disposable = Observable.interval(0, 1, TimeUnit.SECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
return aLong + 1;
}
})
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long count) throws Exception {
Log.d(TAG, "count: " + count);
if (count == 10) {
if (disposable != null) {
disposable.dispose();
}
}
}
});
}
void stop() {
if (disposable != null) {
disposable.dispose();
}
}
上面我们采用了当满足某个条件（count == 10）时候，手动调用了disposable的dispose方法来终止数据继续发射。实际上，RxJava提供了take操作符，可以用来限定要接收的数据数。那么，改写一下：
disposable = Observable.interval(0, 1, TimeUnit.SECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
return aLong + 1;
}
})
.take(10)
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long count) throws Exception {
Log.d(TAG, "count: " + count);
}
});
对于上面的例子，实际上还可以再简单一点，RxJava提供了intervalRange操作符，可以限定要发送的数据个数。再改写一下：
disposable = Observable.intervalRange(0, 10, 1, 1, TimeUnit.SECONDS)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long aLong) throws Exception {
return aLong + 1;
}
})
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long count) throws Exception {
Log.d(TAG, "count: " + count);
}
});
讲了这么多，上面的例子貌似没太多用，毕竟都是简单的轮询计数，也就用来做个UI方面的倒计时有用点。假如，我们需要轮询去做一个耗时操作，比如轮询请求网络呢，这个用RxJava该怎么实现呢？
假设我们的网络请求是从服务器获得某个数，因为是网络请求，所以需要考虑失败的情况。我们伪造一个数据源如下：
private Observable<Integer> getDataFromServer() {
return Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
if (emitter.isDisposed()) {
return;
}
int randomSleep = new Random().nextInt(5);
try {
Thread.sleep(randomSleep * 1000);
} catch (Exception e) {}
if (emitter.isDisposed()) {
return;
}
if (randomSleep % 2 == 0) {
emitter.onError(new Exception("get fake error for " + randomSleep));
return;
}
emitter.onNext(randomSleep);
emitter.onComplete();
}
});
}
第一个版本：定时发送消息，然后收到消息就执行网络请求
结合上面，我们很容易想到，可以利用interval或者intervalRange操作符定时发送消息，接收到消息后就开始执行网络请求。
CompositeDisposable compositeDisposable = new CompositeDisposable();
@Override
public void start() {
compositeDisposable.dispose();
compositeDisposable = new CompositeDisposable();
loopAtFixRate();
//loopSequence();
}
// 嵌套风格loop， 不管实际结果，反正到点了就执行。
private void loopAtFixRate() {
compositeDisposable.add(Observable.interval(0, 5, TimeUnit.SECONDS)
.subscribe(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
Log.d(TAG, "interval: " + aLong);
getData();
}
}));
}
private void getData() {
compositeDisposable.add(getDataFromServer()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "getData: " + integer);
view.showText(integer + "");
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.d(TAG, "getData error " + throwable.getMessage());
view.showText(throwable.getMessage());
}
}));
}
@Override
public void stop() {
if (compositeDisposable != null) {
compositeDisposable.dispose();
}
}
这个版本是最容易想到的，也是最简单的。但是有一些缺点，比如：嵌套风格的Rx代码显得比较丑陋。另外，
每个请求不能按照顺序执行，可能会出现后发的请求先到的情况。如果轮询的任务需要请求顺序执行的话，或者下次轮询的间隔跟请求结果相关联的话，这种方式就不适用。
针对这两种缺点，我们依次试试看。
第二个版本：去掉嵌套的固定请求间隔的loop
利用RxJava的flatMap操作符可以将获取数据的Observable链到原始消息数据源上。这样就不存在嵌套了。OK，先看看下面：
private void loopAtFixRateEx() {
compositeDisposable.add(Observable.interval(0, 5, TimeUnit.SECONDS)
.flatMap(new Function<Long, ObservableSource<Integer>>() {
@Override
public ObservableSource<Integer> apply(Long aLong) throws Exception {
return getDataFromServer();
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer value) throws Exception {
Log.d(TAG, "value: " + value);
view.showText(value + "");
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.e(TAG, "loopAtFixRateEx", throwable);
}
}));
}
上面代码看起来是OK的，但是你一跑起来就会发现从服务端拉取数据失败后上面并不会进行重试。原因就在于flatMap链接了getDataFromServer的数据源，而这个数据源抛出的异常会转移到最外面订阅的onError回调中，并且默认出现error，就会dispose整个数据源。所以也就终止了轮询操作。ok，马上分析怎么解决这个问题。
flatMap有个参数delayErrors，如果传入为true，表示遇到错误后不会立即抛出来，等到所有数据发射完了，或者dispose了之后再抛出来。所以，是不是传入delayErrors为true就解决了。
实际上，想多了，这个会带来新的问题，如果任务一直运行，所有的error都会累积到内存中，会导致内存溢出。另外，我实际测试发现，如果有error产生，我调用了dispose，会导致crash。具体怎么解决，我还没找到办法，所以，用flatMap这个就算了。
第三个版本：采用repeat、结合retry实现轮询
RxJava中的repeat操作符可以在原始数据源发射数据完成后重新订阅数据源，而retry可以在原始数据源产生错误后重新订阅数据源。结合起来就可以在无论是成功还是失败的都能重新执行任务，则实现了轮询请求。再结合delay操作符，实现延迟执行任务。
// 按照顺序loop，意味着第一次结果请求完成后，再考虑下次请求
private void loopSequence() {
Disposable disposable = getDataFromServer()
.doOnSubscribe(new Consumer<Disposable>() {
@Override
public void accept(Disposable disposable) throws Exception {
Log.d(TAG, "loopSequence subscribe");
}
})
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "loopSequence doOnNext: " + integer);
}
})
.doOnError(new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
Log.d(TAG, "loopSequence doOnError: " + throwable.getMessage());
}
})
.delay(5, TimeUnit.SECONDS, true) // 设置delayError为true，表示出现错误的时候也需要延迟5s进行通知，达到无论是请求正常还是请求失败，都是5s后重新订阅，即重新请求。
.subscribeOn(Schedulers.io())
.repeat() // repeat保证请求成功后能够重新订阅。
.retry() // retry保证请求失败后能重新订阅
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
view.showText(integer + "");
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
view.showText(throwable.getMessage());
}
});
compositeDisposable.add(disposable);
}
上面这个例子是对轮询的其中一个场景的示范，实际上你的需求可能会千奇百怪，比如说，有轮询次数限制，并且轮询间隔需要根据轮询的次数做调整，或者由返回的结果来决定下次轮询的时间间隔等。这时候，可能就需要多尝试，多实验才能玩对了。
RxJava要实现一个完全正确可用的轮询，还是需要多测试的。不过，你对RxJava了解的越多，用起来就越爽。
综上，如果是比较简单的轮询，上面哪种方式都可以玩。如果稍微复杂点呢，如果你没办法用RxJava玩对的话，建议就用原始的Handler+postDelay或者Java的Timer来做，先保证做对，再保证做好。
附上类似文章Sample源码：https://github.com/luckyshane/LoopSample
