android - 订阅后一次性对象为空

Question

在下面的代码中，我创建了一个示例来学习使用 Rx 进行函数式编程。我正在尝试将 HandlerThread 处理为可观察的。在 onResume() 中，我订阅了 Single.just observable 以启动 HandlerThread.

对于 SingleObserver 回调，尽管在 onSuccess() 中接收到值，onSubscribe() 中的 Disposable 对象> 始终为 null。

我还发布了 logcat。请看一下，请告诉我为什么 Disposable 对象 d 是 null。

代码:

onResume() {
    this.mMyHandlerThreadInitSingleObs = Single.just(this.getInitializedHandlerThread())
            .map(myHandlerThread->{
                Log.d(TAG_LOG, "BEFORE .start()");
                myHandlerThread.start();
                Log.d(TAG_LOG, "AFTER .start()");

                return this.mMyHandlerThread;
            });
    this.mMyHandlerThreadInitSingleObs
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this.getSingleObserver());
}

private SingleObserver<HandlerThread> getSingleObserver() {
    String TAG_LOG = ActMain.TAG_LOG + "." + "getSingleObserver()";
    return new SingleObserver<HandlerThread>() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.v(TAG_LOG, "[onSubscribe] d: " + d);
        }

        @Override
        public void onSuccess(HandlerThread is) {
            Log.v(TAG_LOG, "[onSuccess] is: " + is);
        }

        @Override
        public void onError(Throwable e) {
            Log.e(TAG_LOG, "[onError]");
        }
    };
}

日志:

2018-12-22 14:56:50.329 12611-12611 V/ActMain: onStart
2018-12-22 14:56:50.332 12611-12611 V/ActMain.MyHandlerThread: constructor called
2018-12-22 14:56:50.333 12611-12611 V/ActMain.getSingleObserver(): [onSubscribe] d: null//<--------------------
2018-12-22 14:56:50.349 12611-12611 D/ActMain.onResume(): BEFORE .start()
2018-12-22 14:56:50.349 12611-12611 D/ActMain.onResume(): AFTER .start()
2018-12-22 14:56:50.350 12611-12630 V/ActMain.MyHandlerThread.onLooperPrepared: ..
2018-12-22 14:56:50.350 12611-12630 D/ActMain.MyHandlerThread.onLooperPrepared: this.getLooper(): Looper (my HandlerThread, tid 416) {2f35ee2}
2018-12-22 14:56:50.363 12611-12630 I/ActMain.MyHandlerThread.onLooperPrepared: [onSubscribe] d: null
2018-12-22 14:56:50.377 12611-12633 D/ActMain.MyHandlerThread.onLooperPrepared.emitter.onComplete():: this.getLooper() initialized: Looper (my HandlerThread, tid 416) {2f35ee2}
2018-12-22 14:56:50.377 12611-12633 I/ActMain.MyHandlerThread.onLooperPrepared: [onComplete]
2018-12-22 14:56:50.425 12611-12611 V/ActMain.getSingleObserver(): [onSuccess] is: Thread[my HandlerThread,5,main]
2018-12-22 14:56:50.514 1700-1724 I/ActivityManager: Displayed com.example.amrbakri.rxhandlerthread_01/.ActMain: +340ms

Answer 1

事实证明，这是一个有趣的问题。来自 RxJava2 的文档 ( Single )

Single behaves similarly to Observable except that it can only emit either a single successful value or an error (there is no "onComplete" notification as there is for an Observable).

The Single class implements the SingleSource base interface and the default consumer type it interacts with is the SingleObserver via the subscribe(SingleObserver) method.

看起来您使用的是 SingleObserver 而不是 DisposableSingleObserver。

文档提到:

Note that by design, subscriptions via subscribe(SingleObserver) can't be disposed from the outside (hence the void return of the subscribe(SingleObserver) method) and it is the responsibility of the implementor of the SingleObserver to allow this to happen. RxJava supports such usage with the standard DisposableSingleObserver instance. For convenience, the subscribeWith(SingleObserver) method is provided as well to allow working with a SingleObserver (or subclass) instance to be applied with in a fluent manner (such as in the example above).

所以尝试这样做:

Disposable d = Single.just("Hello World")
    .delay(10, TimeUnit.SECONDS, Schedulers.io())
    .subscribeWith(new DisposableSingleObserver<String>() {
        @Override
        public void onStart() {
            System.out.println("Started");
        }

        @Override
        public void onSuccess(String value) {
            System.out.println("Success: " + value);
        }

        @Override
        public void onError(Throwable error) {
            error.printStackTrace();
        }
    });