android - observeOn() 影响 doOnTerminate() 调用

Question

我有以下代码

Observable.just(10)
            .doOnTerminate(() -> Log.d("LOG", "ON TERMINATE"))
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(result -> {
                int a = result / 0; // force an exception
            }, error -> {
                Log.d("LOG", "ERROR");
                error.printStackTrace();
            });

这将给出输出:

日志:终止

日志:错误

但是如果我稍微修改成

Observable.just(10)
             .subscribeOn(Schedulers.newThread())
             .observeOn(AndroidSchedulers.mainThread())
             .doOnTerminate(() -> Log.d("LOG", "ON TERMINATE"))
             .subscribe(result -> {
                 int a = result / 0; // force an exception
             }, error -> {
                 Log.d("LOG", "ERROR");
                 error.printStackTrace();
             });

它只会给日志:错误

为什么 doOnTerminate 函数没有在第二个代码上被调用？

Answer 1

我认为这是因为应该在哪里调用 doOnTerminate。在第一种情况下，它在新线程上被调用。所以事件如下:

+-newThread-------------------------------------------------------------------------+
| emission      ----- 10 -- |(completion because only one item was emitted)         |
| doOnTerminate            call(reason for calling is that emissions completed)     |
+-mainThread------------------------------------------------------------------------+
| subscriber    ----- 10(throw div/0)---X                                           |
+-----------------------------------------------------------------------------------+

如果你看看如何just operator是在引擎盖下实现的，您会看到它在 onNext 之后立即调用 onCompleted。

当你将 doOnTerminate 移动到主节点时

+-newThread-------------------------------------------------------------------------+
| emission      ----- 10 -- |(completion because only one item was emitted)         |
+-mainThread------------------------------------------------------------------------+
| doOnTerminate                                                        (not called) |
| subscriber    ----- 10(throw div/0)->X                                            |
+-----------------------------------------------------------------------------------+

不调用 doOnTerminate 的原因是错误发生在订阅者的 onNext 部分。在这种情况下，它是 LambdaObserver当检测到 onNext 中的错误时直接调用 onError(参见链接)。

---

我的测试代码:

    final long[] start = {0};
    Observable.just(10)
            .subscribeOn(Schedulers.newThread())
            .doOnNext(integer -> {
                start[0] = System.currentTimeMillis();
                message("onNext", start[0]);
            })
            .doOnTerminate(() -> message("doOnTerminate", start[0]))
            .doFinally(() -> message("doFinally", start[0]))
            .doAfterTerminate(() -> message("doAfterTerminate", start[0]))
            .observeOn(AndroidSchedulers.mainThread())
            .doOnTerminate(() -> message("doOnTerminate", start[0]))
            .doFinally(() -> message("doFinally", start[0]))
            .doAfterTerminate(() -> message("doAfterTerminate", start[0]))
            .subscribe(integer -> {
                        message("Next", start[0]);
                        int a = integer / 0;
                    },
                    throwable -> message("ERROR", start[0])
            );

     void message(String message, long start) {
            Log.d("LOG", message + " " + Thread.currentThread().getName() + " " + (System.currentTimeMillis() - start));
     }

它的输出:

D/LOG: onNext RxNewThreadScheduler-1 0
D/LOG: doOnTerminate RxNewThreadScheduler-1 0
D/LOG: doAfterTerminate RxNewThreadScheduler-1 0
D/LOG: doFinally RxNewThreadScheduler-1 0
D/LOG: Next main 16
D/LOG: doFinally main 16
D/LOG: ERROR main 16

---

解决方案

为了避免这种情况并同时避免丢失取消订阅(处置)事件，我将使用 doFinally 运算符。

所有这些都是基于 RxJava 2.0.4 版本。