如何将异步任务/ rx java 代码转换为 rxjava2？

Question

我尝试将下面的 AsyncTask 代码转换为 Rxjava2，但显然 Rxjava2 不处理空值，因此我的应用程序崩溃了。这是我的 AsyncTask 代码：

new AsyncTask<Void, Void, Void>() {
            @Override
            protected Void doInBackground(Void... params) {
                Set<Map.Entry<String, Participant>> entries = pool.entrySet();
                for (Map.Entry<String, Participant> entry : entries) {
                    Participant participant = entry.getValue();
                    participant.release();
                }
                return null;
            }

            @Override
            protected void onPostExecute(Void aVoid) {
                cb.event(new Spin.Event<Void>());
            }
        }.execute();

这里是转换后的代码 Rxjava（不是 Rxjava2）：

 Observable.defer(new Func0<Observable<Void>>() {
        @Override
        public Observable<Void> call() {
            Set<Map.Entry<String, Participant>> entries = pool.entrySet();
            for (Map.Entry<String, Participant> entry : entries) {
                Participant participant = entry.getValue();
                participant.release();
            }
            return Observable.just(null);
        }
    }).doOnCompleted(new Action0() {
        @Override
        public void call() {
            cb.event(new Spin.Event<Void>());
        }
    })
    .subscribeOn(Schedulers.computation())
    .subscribe();

将其转换为 Rxjava 且不会在返回 null 时崩溃的最佳方法是什么。另外，.execute() 对 Rxjava2 有何影响？不确定这是否适用于 Rxjava?

这是崩溃日志：

FATAL EXCEPTION: RxComputationThreadPool-3

                                                                           io.reactivex.exceptions.OnErrorNotImplementedException: null ObservableSource supplied
                                                                               at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
                                                                               at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
                                                                               at io.reactivex.internal.observers.LambdaObserver.onError(LambdaObserver.java:74)
                                                                               at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.onError(ObservableSubscribeOn.java:63)
                                                                               at io.reactivex.internal.disposables.EmptyDisposable.error(EmptyDisposable.java:63)
                                                                               at io.reactivex.internal.operators.observable.ObservableDefer.subscribeActual(ObservableDefer.java:35)
                                                                               at io.reactivex.Observable.subscribe(Observable.java:10842)
                                                                               at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
                                                                               at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38)
                                                                               at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26)
                                                                               at java.util.concurrent.FutureTask.run(FutureTask.java:237)
                                                                               at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access1(ScheduledThreadPoolExecutor.java:154)
                                                                               at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:269)
                                                                               at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1113)
                                                                               at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:588)
                                                                               at java.lang.Thread.run(Thread.java:818)
                                                                            Caused by: java.lang.NullPointerException: null ObservableSource supplied
                                                                               at io.reactivex.internal.functions.ObjectHelper.requireNonNull(ObjectHelper.java:39)
                                                                               at io.reactivex.internal.operators.observable.ObservableDefer.subscribeActual(ObservableDefer.java:32)
                                                                               at io.reactivex.Observable.subscribe(Observable.java:10842) 
                                                                               at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96) 
                                                                               at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38) 
                                                                               at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26) 
                                                                               at java.util.concurrent.FutureTask.run(FutureTask.java:237) 
                                                                               at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access1(ScheduledThreadPoolExecutor.java:154) 
                                                                               at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:269) 
                                                                               at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1113) 
                                                                               at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:588) 
                                                                               at java.lang.Thread.run(Thread.java:818)

Answer 1

Observable.defer(new Callable<ObservableSource<?>>() {
            @Override
            public ObservableSource<?> call() throws Exception {
                Set<Map.Entry<String, Participant>> entries = pool.entrySet();
                for (Map.Entry<String, Participant> entry : entries) {
                   Participant participant = entry.getValue();
                   participant.release();
                }
                return Completable.complete().toObservable();
            }
        }).doOnComplete(new Action() {
            @Override
            public void run() throws Exception {
                Log.d("Complete", "Complete");
            }
        })
            .subscribeOn(Schedulers.computation())
            .observeOn(AndroidSchedulers.mainThread()).subscribe();

此代码也可以。调用 subscribe() 方法将启动作业。

Answer 2

由于您没有 post 返回主线程的值，您可以使用 Completable:

Completable.fromAction(() -> {
    Set<Map.Entry<String, Participant>> entries = pool.entrySet();
    for (Map.Entry<String, Participant> entry : entries) {
        Participant participant = entry.getValue();
        participant.release();
    }
})
.subscribeOn(Schedulers.computation())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
    () -> {
         cb.event(new Spin.Event<Void>());
    }, 
    error -> { /* show error toast */ }
);

Answer 3

Observable.defer(new Callable<ObservableSource<?>>() {

 //This method is replacing doInBackground
        @Override
        public ObservableSource<?> call() throws Exception {  
            Set<Map.Entry<String, Participant>> entries = pool.entrySet();
            for (Map.Entry<String, Participant> entry : entries) {
               Participant participant = entry.getValue();
               participant.release();
            }
            return Completable.complete().toObservable();
        }
    }).doOnComplete(new Action() {
     //This is onPostExecute
        @Override 
        public void run() throws Exception {
            Log.d("Complete", "Complete");
        }
    })
        .subscribeOn(Schedulers.computation())
        .observeOn(AndroidSchedulers.mainThread()).subscribe()

如何将异步任务/ rx java 代码转换为 rxjava2？

How do I convert async task/ rx java code to rxjava2?

android

android-asynctask

rx-android

rx-java2