RxJava2使用过程中遇到的坑

[qqiabc521]

1. 初始化RxJava回收执行周期

背景：RxJava开启一个循环线程在后台默默回收Publisher，默认是1秒循环一次，但是这样比较耗费cpu，纯净后台检测时，cpu唤醒率无法达到标准。
解决方案：设置垃圾检测回收周期为5分钟。

初始化RxJava回收执行的周期
@see io.reactivex.internal.schedulers.SchedulerPoolFactory
PURGE_ENABLED_KEY与PURGE_PERIOD_SECONDS_KEY非public属性参数，外界不能直接访问，需要通过转换，在项目代码中新建io.reactivex.internal.schedulers包

package io.reactivex.internal.schedulers;
public class SchedulerPoolConfig {
public static final String PURGE_ENABLED_KEY = SchedulerPoolFactory.PURGE_ENABLED_KEY;
public static final String PURGE_PERIOD_SECONDS_KEY = SchedulerPoolFactory.PURGE_PERIOD_SECONDS_KEY;
}

private static void initRxPurgeProperties() {
System.setProperty(SchedulerPoolConfig.PURGE_ENABLED_KEY, "true");
System.setProperty(SchedulerPoolConfig.PURGE_PERIOD_SECONDS_KEY, "30");
}

2. RxJava2 取消订阅后，抛出的异常无法捕获，导致程序崩溃

最近在项目中，就遇到了host解析失败导致程序崩溃的情况：
io.reactivex.exceptions.UndeliverableException: java.net.UnknownHostException: Unable to resolve host "api-test.zhangyuke.com": No address associated with hostname

当时遇到这个问题，百思不得其解。因为Rxjava有一个很大的优势，就是可以帮助使用者捕获异常，而且在对应了异常捕获中，已经有针对性的对UnknownHostException异常做了处理，并给用户以提示。后来在跟踪这个错误日志发现， java.net.UnknownHostException仅仅是被包裹的异常，最外面的一层异常是io.reactivex.exceptions.UndeliverableException，UndeliverableException从字面意思看，意思是不可送达的异常。通过源码跟踪，发现抛出io.reactivex.exceptions.UndeliverableException唯一地方是RxJavaPlugins类：

public static void onError(@nonnull Throwable error) {
Consumer<? super Throwable> f = errorHandler;

if (error == null) {
error = new NullPointerException("onError called with null. Null values are generally not allowed in 2.x operators and sources.");
} else {
if (!isBug(error)) {
error = new UndeliverableException(error);
}
}
if (f != null) {
try {
f.accept(error);
return;
} catch (Throwable e) {
// Exceptions.throwIfFatal(e); TODO decide
e.printStackTrace(); // NOPMD
uncaught(e);
}
}
error.printStackTrace(); // NOPMD
uncaught(error);
}

背景：RxJava2的一个重要的设计理念是：不吃掉任何一个异常。产生的问题是，当RxJava2“downStream”取消订阅后，“upStream”仍有可能抛出异常，这时由于已经取消订阅，“downStream”无法处理异常，此时的异常无人处理，便会导致程序崩溃。

示例代码：
mDisposable = Flowable("token")
.map(new Function<String, Object>() {
@OverRide
public List apply(String token) throws Exception {
//...
//code A
throw new InterceptorIOException("io error");//throw exception when mDisposable.isCanceled()
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer() {
@OverRide
public void accept(Objeect data) throws Exception {
// do something when success
}
}, new Consumer() {
@OverRide
public void accept(Throwable throwable) throws Exception {
//do something when error!
//code B
//注意：当mDisposable.isCanceled()时抛出的异常，这里不会补货，因为已经取消了订阅
}
});

上述代码code A出抛出的异常code B处无法捕获，便会导致程序崩溃。

解决方案：在Application设置RxJavaPlugin的ErrorHandler
/**
*RxJava2 当取消订阅后(dispose())，RxJava抛出的异常后续无法接收(此时后台线程仍在跑，可能会抛出IO等异常),全部由RxJavaPlugin接收，需要提前设置ErrorHandler
*详情：http://engineering.rallyhealth.com/mobile/rxjava/reactive/2017/03/15/migrating-to-rxjava-2.html#Error Handling
*/

private void setRxJavaErrorHandler() {
RxJavaPlugins.setErrorHandler(new Consumer() {
@OverRide
public void accept(Throwable throwable) throws Exception {
throwable.printStackTrace();;
AppLog.d("ljj", "throw : "+throwable.getMessage());
}
});
}