关于android:RxJava-堆栈异常信息显示不全怎么搞

原文：徐公

景象

大家好，明天为大家带来的是 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/R…

第一步，引入依赖库

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…

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

最初

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