本文首發我的部落格，github 地址

大家好，我是徐公，今天為大家帶來的是 RxJava 的一個血案，一行程式碼 return null 引發的。

前陣子，組內的同事反饋說 RxJava 在 debug 包 crash 了，捕獲到的異常資訊不全。（即我們捕獲到的堆疊沒有包含我們自己程式碼，都是一些系統或者 RxJava 框架的程式碼）

典型的一些 error 資訊如下：

可以看到，上面的 Error 堆疊資訊中，它並沒有給出這個 Error 在實際專案中的呼叫路徑。可以看到，報錯的堆疊，提供的有效資訊較少，我們只能知道是由於 callable.call() 這裡返回了 Null，導致出錯。卻不能判斷 callable 是哪裡建立的，這時候我們只能結合紀錄檔上下文，判斷當前之前的程式碼大概在哪裡，再逐步排查。

public final class ObservableFromCallable<T> extends Observable<T> implements Callable<T> {
  

    @Override
    public void subscribeActual(Observer<? super T> observer) {
        DeferredScalarDisposable<T> d = new DeferredScalarDisposable<T>(observer);
        observer.onSubscribe(d);
        if (d.isDisposed()) {
            return;
        }
        T value;
        try {
            // callable.call()  這裡返回了 Null，並傳遞給了 RxJavaPlugins 的 errorHandler
            value = ObjectHelper.requireNonNull(callable.call(), "Callable returned null");
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            if (!d.isDisposed()) {
                observer.onError(e);
            } else {
                RxJavaPlugins.onError(e);
            }
            return;
        }
        d.complete(value);
    }

}

一頓操作猛如虎，很多，我們結合一些讓下文紀錄檔，發現是這裡返回了 null，導致出錯

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

/**
 * 建立一個rx的子執行緒任務Observable
 */
private fun <T> backgroundTask(callable: Callable<T>?): Observable<T>? {
    return Observable.fromCallable(callable)
            .compose(IOMain())
}

如果遇到 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

    fun hookRxFromCallable() {
//        DexposedBridge.findAndHookMethod(ObservableFromCallable::class.java, "subscribeActual", Observer::class.java, RxMethodHook())
        DexposedBridge.findAndHookMethod(
            Observable::class.java,
            "fromCallable",
            Callable::class.java,
            object : XC_MethodHook() {
                override fun beforeHookedMethod(param: MethodHookParam?) {
                    super.beforeHookedMethod(param)
                    val args = param?.args
                    args ?: return

                    val callable = args[0] as Callable<*>
                    args[0] = MyCallable(callable = callable)

                }

                override fun afterHookedMethod(param: MethodHookParam?) {
                    super.afterHookedMethod(param)
                }
            })
    }

    class MyCallable(private val callable: Callable<*>) : Callable<Any> {

        private val TAG = "RxJavaHookActivity"
        val buildStackTrace: String?

        init {
            buildStackTrace = Rx2Utils.buildStackTrace()
        }

        override fun call(): Any {
            Log.i(TAG, "call: ")
            val call = callable.call()
            if (call == null) {
                Log.e(TAG, "call should not return null: buildStackTrace is $buildStackTrace")
            }
            return call
        }

    }

再次執行我們的程式碼

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

可以看到，當我們的 Callable 返回為 empty 的時候，這時候報錯的資訊會含有我們專案的程式碼， perfect。

RxJavaExtensions

最近，在 Github 上面發現了這一個框架，它也可以幫助我們解決 RxJava 異常過程中資訊不全的問題。它的基本使用如下：

使用

https://github.com/akarnokd/RxJavaExtensions

第一步，引入依賴庫

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

第二步：先啟用錯誤追蹤：

RxJavaAssemblyTracking.enable();

第三步：在丟擲異常的異常，列印堆疊

    /**
     * 設定全域性的 onErrorHandler。
     */
    fun setRxOnErrorHandler() {
        RxJavaPlugins.setErrorHandler { throwable: Throwable ->
            val assembled = RxJavaAssemblyException.find(throwable)
            if (assembled != null) {
                Log.e(TAG, assembled.stacktrace())
            }
            throwable.printStackTrace()
            Log.e(TAG, "setRxOnErrorHandler: throwable is $throwable")
        }
    }

原理

RxJavaAssemblyTracking.enable();

public static void enable() {
    if (lock.compareAndSet(false, true)) {
 
        // 省略了若干方法

        RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {
            @Override
            public Observable apply(Observable f) throws Exception {
                if (f instanceof Callable) {
                    if (f instanceof ScalarCallable) {
                        return new ObservableOnAssemblyScalarCallable(f);
                    }
                    return new ObservableOnAssemblyCallable(f);
                }
                return new ObservableOnAssembly(f);
            }
        });

     
        lock.set(false);
    }
}

可以看到，它呼叫了 RxJavaPlugins.setOnObservableAssembly 方法，設定了 RxJavaPlugins onObservableAssembly 變數

而我們上面提到的 Observable#fromCallable 方法，它裡面會呼叫 RxJavaPlugins.onAssembly 方法，當我們的 onObservableAssembly 不為 null 的時候，會呼叫 apply 方法進行轉換。

public static <T> Observable<T> fromCallable(Callable<? extends T> supplier) {
    ObjectHelper.requireNonNull(supplier, "supplier is null");
    return RxJavaPlugins.onAssembly(new ObservableFromCallable<T>(supplier));
}
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
    Function<? super Observable, ? extends Observable> f = onObservableAssembly;
    if (f != null) {
        return apply(f, source);
    }
    return source;
}

因此，即當我們設定了 RxJavaAssemblyTracking.enable()， Observable#fromCallable 傳遞進來的 supplier，最終會包裹一層，可能是 ObservableOnAssemblyScalarCallable，ObservableOnAssemblyCallable，ObservableOnAssembly。典型的裝飾者模式應用，這裡不得不說，RxJava 對外提供的這個點，設計得真巧妙，可以很方便我們做一些 hook。

我們就以 ObservableOnAssemblyCallable 看一下

final class ObservableOnAssemblyCallable<T> extends Observable<T> implements Callable<T> {

    final ObservableSource<T> source;

    // 將在哪裡建立的 Callable 的堆疊資訊儲存下來
    final RxJavaAssemblyException assembled;

    ObservableOnAssemblyCallable(ObservableSource<T> source) {
        this.source = source;
        this.assembled = new RxJavaAssemblyException();
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        source.subscribe(new OnAssemblyObserver<T>(observer, assembled));
    }

    @SuppressWarnings("unchecked")
    @Override
    public T call() throws Exception {
        try {
            return ((Callable<T>)source).call();
        } catch (Exception ex) {
            Exceptions.throwIfFatal(ex);
            throw (Exception)assembled.appendLast(ex);
        }
    }
}

public final class RxJavaAssemblyException extends RuntimeException {

    private static final long serialVersionUID = -6757520270386306081L;

    final String stacktrace;

    public RxJavaAssemblyException() {
        this.stacktrace = buildStackTrace();
    }
 }

可以看到，他是直接在 ObservableOnAssemblyCallable 的構造方法的時候，直接將 Callable 的堆疊資訊儲存下來，類為 RxJavaAssemblyException。

而當 error 報錯的時候，呼叫 RxJavaAssemblyException.find(throwable) 方式，判斷是不是 RxJavaAssemblyException，是的話，直接返回。

public static RxJavaAssemblyException find(Throwable ex) {
    Set<Throwable> memory = new HashSet<Throwable>();
    while (ex != null) {
        if (ex instanceof RxJavaAssemblyException) {
            return (RxJavaAssemblyException)ex;
        }

        if (memory.add(ex)) {
            ex = ex.getCause();
        } else {
            return null;
        }
    }
    return null;
}

到這裡，RxJavaAssemblyTracking 能將 error 資訊完整列印出來的流程已經講明白了，其實就是在建立 Callable 的時候，採用一個包裝類，在建構函式的時候，將 error 資訊報錯下來，等到出錯的時候，再將 error 資訊，替換成儲存下來的 error資訊。

我們的自定義 Hook 也是利用這種思路，提前將 callable 建立的堆疊暴露下來，換湯不換藥。

一些思考

上述的方案我們一般不會帶到線上，為什麼呢？因為對於每一個 callable，我們需要提前儲存堆疊，而獲取堆疊是耗時的。那有沒有什麼方法呢？

如果專案有接入 Matrix 的話，可以考慮借用 Matrix trace 的思想，因為在方法前後插入 AppMethodBeat#i 和 AppMethodBeat#o 這樣當我們執行方法的時候，因為插樁了，我們可以方便得獲取到方法執行耗時，以及方法的呼叫棧。

// 第一步：需要在合適的實際先生成 beginRecord
AppMethodBeat.IndexRecord  beginRecord = AppMethodBeat.getInstance().maskIndex("AnrTracer#dispatchBegin");
// 第二步：方法的呼叫棧資訊在 data 裡面
long[] data = AppMethodBeat.getInstance().copyData(beginRecord);
第三步：
將 data 轉化為我們想要的 stack（初步看了程式碼，需要我們修改 trace 的程式碼）

參考資料

rxjava-2-doesnt-tell-the-error-line

how-to-log-a-stacktrace-of-all-exceptions-of-rxjava2

RxJava 異常時堆疊顯示不正確？解決方法都在這裡

第一種方案，自定義 Hook 解決

RxJavaExtensions

使用

原理

一些思考

參考資料

推薦閱讀