原文：徐公

现象

大家好，今天为大家带来的是 RxJava 的一个血案，一行代码 return null 引发的。

前阵子，组内的同事反馈说 RxJava 在 debug 包 crash 了，捕获到的异常信息不全。（即我们捕获到的堆栈没有包含我们自己代码，都是一些系统或者 RxJava 框架的代码）

典型的一些 error 信息如下：

 1 io.reactivex.exceptions.OnErrorNotImplementedException: The exception was not handled due to missing onError handler in the subscribe() method call. Further reading: https://github.com/ReactiveX/RxJava/wiki/Error-Handling | java.lang.NullPointerException: Callable returned null
 2     at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
 3     at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
 4     at io.reactivex.internal.observers.LambdaObserver.onError(LambdaObserver.java:77)
 5     at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.checkTerminated(ObservableObserveOn.java:281)
 6     at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:172)
 7     at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:255)
 8     at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:124)
 9     at android.os.Handler.handleCallback(Handler.java:883)
10     at android.os.Handler.dispatchMessage(Handler.java:100)
11     at android.os.Looper.loop(Looper.java:214)
12     at android.app.ActivityThread.main(ActivityThread.java:7682)
13     at java.lang.reflect.Method.invoke(Native Method)
14     at com.android.internal.os.RuntimeInit$MethodAndArgsCaller.run(RuntimeInit.java:516)
15     at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:950)
16 Caused by: java.lang.NullPointerException: Callable returned null
17     at io.reactivex.internal.functions.ObjectHelper.requireNonNull(ObjectHelper.java:39)
18     at io.reactivex.internal.operators.observable.ObservableFromCallable.subscribeActual(ObservableFromCallable.java:43)
19     at io.reactivex.Observable.subscribe(Observable.java:12267)
20     at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
21     at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:578)
22     at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)
23     at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:57)
24     at java.util.concurrent.FutureTask.run(FutureTask.java:266)
25     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:301)
26     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)
27     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)
28     at java.lang.Thread.run(Thread.java:919)

可以看到，上面的 Error 堆栈信息中，它并没有给出这个 Error 在实际项目中的调用路径。可以看到，报错的堆栈，提供的有效信息较少， 我们只能知道是由于 callable.call() 这里返回了 Null，导致出错。却不能判断 callable 是哪里创建的，这时候我们只能结合日志上下文，判断当前之前的代码大概在哪里，再逐步排查。

 1        public final class ObservableFromCallable<T> extends Observable<T> implements Callable<T> {
 2
 3
 4        @Override
 5        public void subscribeActual(Observer<? super T> observer) {
 6        DeferredScalarDisposable<T> d = new DeferredScalarDisposable<T>(observer);
 7        observer.onSubscribe(d);
 8        if (d.isDisposed()) {
 9            return;
10        }
11        T value;
12        try {
13            // callable.call()  这里返回了 Null，并传递给了 RxJavaPlugins 的 errorHandler
14            value = ObjectHelper.requireNonNull(callable.call(), "Callable returned null");
15        } catch (Throwable e) {
16            Exceptions.throwIfFatal(e);
17            if (!d.isDisposed()) {
18                observer.onError(e);
19            } else {
20                RxJavaPlugins.onError(e);
21            }
22            return;
23        }
24        d.complete(value);
25       }
26
27      }

一顿操作猛如虎，很多，我们结合一些上下文日志，发现是这里返回了 null，导致出错

1    backgroundTask(Callable<Any> {
2    Log.i(TAG, "btn_rx_task: ")
3    Thread.sleep(30)
4    return@Callable null
5    })?.subscribe()

1/**
2 * 创建一个rx的子线程任务Observable
3 */
4    private fun <T> backgroundTask(callable: Callable<T>?): Observable<T>? {
5    return Observable.fromCallable(callable)
6            .compose(IOMain())
7    }

如果遇到 callable 比较多的情况下，这时候 一个个排查 callable，估计搞到你吐血。

那有没有什么较好的方法，比如做一些监控？完整打印堆栈信息。

第一种方案，自定义 Hook 解决

首先，我们先来想一下，什么是堆栈？

在我的理解里面，堆栈是用来储存我们程序当前执行的信息。在 Java 当中，我们通过 java.lang.Thread#getStackTrace 可以拿到当前线程的堆栈信息，注意是当前线程的堆栈。

而 RxJava 抛出异常的地方，是在执行 Callable#call 方法中，它打印的自然是 Callable#call 的方法调用栈，而如果 Callable#call 的调用线程跟 callable 的创建线程不一致，那肯定拿不到 创建 callable 时候的堆栈。

而我们实际上需要知道的是 callable 创建的地方，对应到我们我们项目报错的地方，那自然是 Observable.fromCallable 方法的调用栈。

这时候，我们可以采用 Hook 的方式，来 Hook 我们的代码

为了方便，我们这里采用了 wenshu 大神的 Hook 框架, github, 想自己手动去 Hook 的，可以看一下我两年前写的文章 Android Hook 机制之简单实战，里面有介绍介绍一些常用的 Hook 手段。

很快，我们写出了如下代码，对 Observable#fromCallable 方法进行 hook

 1    fun hookRxFromCallable() {
 2//        DexposedBridge.findAndHookMethod(ObservableFromCallable::class.java, "subscribeActual", Observer::class.java, RxMethodHook())
 3        DexposedBridge.findAndHookMethod(
 4            Observable::class.java,
 5            "fromCallable",
 6            Callable::class.java,
 7            object : XC_MethodHook() {
 8                override fun beforeHookedMethod(param: MethodHookParam?) {
 9                    super.beforeHookedMethod(param)
10                    val args = param?.args
11                    args ?: return
12
13                    val callable = args[0] as Callable<*>
14                    args[0] = MyCallable(callable = callable)
15
16                }
17
18                override fun afterHookedMethod(param: MethodHookParam?) {
19                    super.afterHookedMethod(param)
20                }
21            })
22    }
23
24    class MyCallable(private val callable: Callable<*>) : Callable<Any> {
25
26        private val TAG = "RxJavaHookActivity"
27        val buildStackTrace: String?
28
29        init {
30            buildStackTrace = Rx2Utils.buildStackTrace()
31        }
32
33        override fun call(): Any {
34            Log.i(TAG, "call: ")
35            val call = callable.call()
36            if (call == null) {
37                Log.e(TAG, "call should not return null: buildStackTrace is $buildStackTrace")
38            }
39            return call
40        }
41
42    }

再次执行我们的代码

1    backgroundTask(Callable<Any> {
2    Log.i(TAG, "btn_rx_task: ")
3    Thread.sleep(30)
4    return@Callable null
5    })?.subscribe()

可以看到，当我们的 Callable 返回为 empty 的时候，这时候报错的信息会含有我们项目的代码， perfect。

RxJavaExtensions

最近，在 Github 上面发现了这一个框架，它也可以帮助我们解决 RxJava 异常过程中信息不全的问题。它的基本使用如下：

使用

https://github.com/akarnokd/RxJavaExtensions

第一步，引入依赖库

1dependencies {
2    implementation "com.github.akarnokd:rxjava2-extensions:0.20.10"
3}

第二步：先启用错误追踪：

1RxJavaAssemblyTracking.enable();

第三步：在抛出异常的异常，打印堆栈

 1    /**
 2     * 设置全局的 onErrorHandler。
 3     */
 4    fun setRxOnErrorHandler() {
 5        RxJavaPlugins.setErrorHandler { throwable: Throwable ->
 6            val assembled = RxJavaAssemblyException.find(throwable)
 7            if (assembled != null) {
 8                Log.e(TAG, assembled.stacktrace())
 9        }
10            throwable.printStackTrace()
11            Log.e(TAG, "setRxOnErrorHandler: throwable is $throwable")
12        }
13        }

原理

RxJavaAssemblyTracking.enable();

 1public static void enable() {
 2    if (lock.compareAndSet(false, true)) {
 3
 4        // 省略了若干方法
 5
 6        RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {
 7            @Override
 8            public Observable apply(Observable f) throws Exception {
 9                if (f instanceof Callable) {
10                    if (f instanceof ScalarCallable) {
11                        return new ObservableOnAssemblyScalarCallable(f);
12                    }
13                    return new ObservableOnAssemblyCallable(f);
14                }
15                return new ObservableOnAssembly(f);
16            }
17        });
18
19
20        lock.set(false);
21    }
22    }

可以看到，它调用了 RxJavaPlugins.setOnObservableAssembly 方法，设置了 RxJavaPlugins onObservableAssembly 变量

而我们上面提到的 Observable#fromCallable 方法，它里面会调用 RxJavaPlugins.onAssembly 方法，当我们的 onObservableAssembly 不为 null 的时候，会调用 apply 方法进行转换。

 1     public static <T> Observable<T> fromCallable(Callable<? extends T> supplier) {
 2    ObjectHelper.requireNonNull(supplier, "supplier is null");
 3    return RxJavaPlugins.onAssembly(new ObservableFromCallable<T>(supplier));
 4    }
 5    public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
 6    Function<? super Observable, ? extends Observable> f = onObservableAssembly;
 7    if (f != null) {
 8        return apply(f, source);
 9    }
10    return source;
11    }

因此，即当我们设置了 RxJavaAssemblyTracking.enable()， Observable#fromCallable 传递进来的 supplier，最终会包裹一层，可能是 ObservableOnAssemblyScalarCallable，ObservableOnAssemblyCallable，ObservableOnAssembly。典型的装饰者模式应用，这里不得不说，RxJava 对外提供的这个点，设计得真巧妙，可以很方便我们做一些 hook。

我们就以 ObservableOnAssemblyCallable 看一下

 1   final class ObservableOnAssemblyCallable<T> extends Observable<T> implements Callable<T> {
 2
 3    final ObservableSource<T> source;
 4
 5    // 将在哪里创建的 Callable 的堆栈信息保存下来
 6    final RxJavaAssemblyException assembled;
 7
 8    ObservableOnAssemblyCallable(ObservableSource<T> source) {
 9        this.source = source;
10        this.assembled = new RxJavaAssemblyException();
11    }
12
13    @Override
14    protected void subscribeActual(Observer<? super T> observer) {
15        source.subscribe(new OnAssemblyObserver<T>(observer, assembled));
16    }
17
18    @SuppressWarnings("unchecked")
19    @Override
20    public T call() throws Exception {
21        try {
22            return ((Callable<T>)source).call();
23        } catch (Exception ex) {
24            Exceptions.throwIfFatal(ex);
25            throw (Exception)assembled.appendLast(ex);
26    }
27    }
28    }
29
30   public final class RxJavaAssemblyException extends RuntimeException {
31
32    private static final long serialVersionUID = -6757520270386306081L;
33
34    final String stacktrace;
35
36    public RxJavaAssemblyException() {
37        this.stacktrace = buildStackTrace();
38    }
39    }

可以看到，他是直接在 ObservableOnAssemblyCallable 的构造方法的时候，直接将 Callable 的堆栈信息保存下来，类为 RxJavaAssemblyException。

而当 error 报错的时候，调用 RxJavaAssemblyException.find(throwable) 方式，判断是不是 RxJavaAssemblyException，是的话，直接返回。

 1    public static RxJavaAssemblyException find(Throwable ex) {
 2    Set<Throwable> memory = new HashSet<Throwable>();
 3    while (ex != null) {
 4        if (ex instanceof RxJavaAssemblyException) {
 5            return (RxJavaAssemblyException)ex;
 6        }
 7
 8        if (memory.add(ex)) {
 9            ex = ex.getCause();
10        } else {
11            return null;
12    }
13    }
14    return null;
15    }

到这里，RxJavaAssemblyTracking 能将 error 信息完整打印出来的流程已经讲明白了，其实就是在创建 Callable 的时候，采用一个包装类，在构造函数的时候，将 error 信息报错下来，等到出错的时候，再将 error 信息，替换成保存下来的 error信息。

我们的自定义 Hook 也是利用这种思路，提前将 callable 创建的堆栈暴露下来，换汤不换药。

一些思考

上述的方案我们一般不会带到线上，为什么呢？因为对于每一个 callable，我们需要提前保存堆栈，而获取堆栈是耗时的。那有没有什么方法呢？

如果项目有接入 Matrix 的话，可以考虑借用 Matrix trace 的思想，因为在方法前后插入 AppMethodBeat#i 和 AppMethodBeat#o 这样当我们执行方法的时候，因为插桩了，我们可以方便得获取到方法执行耗时，以及方法的调用栈。

1// 第一步：需要在合适的实际先生成 beginRecord
2AppMethodBeat.IndexRecord  beginRecord = AppMethodBeat.getInstance().maskIndex("AnrTracer#dispatchBegin");
3// 第二步：方法的调用栈信息在 data 里面
4 long[] data = AppMethodBeat.getInstance().copyData(beginRecord);
5第三步：
6将 data 转化为我们想要的 stack（初步看了代码，需要我们修改 trace 的代码）

github:https://github.com/gdutxiaoxu/HookDemo

参考资料
rxjava-2-doesnt-tell-the-error-line
how-to-log-a-stacktrace-of-all-exceptions-of-rxjava2

最后

喜欢的小伙伴别忘了点赞收藏加关注呀~