java - RxJava:在另一个线程上消耗Observable的结果

我需要在后台线程上执行一些工作，然后在主线程上交付结果。我执行以下操作：

Observable.just(new Object()).subscribeOn(Schedulers.newThread()).subscribeWith(new DisposableObserver<Object>() {
            @Override
            public void onNext(Object s) {

                try {
                    doSomething()
                    Observable.just(new Object())
                            .observeOn(AndroidSchedulers.mainThread())
                            .subscribe(new Observer<Object>() {
                                @Override
                                public void onSubscribe(Disposable d) {

                                }

                                @Override
                                public void onNext(Object o) {
                                    completion.deliverResult()
                                    onComplete();
                                }

                                @Override
                                public void onError(Throwable e) {

                                }

                                @Override
                                public void onComplete() {

                                }
                            });
                } catch (DriverException e) {
                    badThingsHappened()
                    onError(e);
                }
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {

            }
        });

但是我不喜欢这段代码，它看起来很复杂并且没有使用很多东西。

有没有办法使它更优雅？

最佳答案

您可以用几行代码来完成所需的操作。

在此代码段中，我将从Callable创建一个Observable对象。计算将在IO线程上完成，但结果（或最终出现错误）将在Android主线程上观察到

void execute() {
    // here you get your observable (api, db, other repositories). I'll create a simple String observable
    Observable<String> observable = Observable.fromCallable(
            () -> "This method will be executed on IO Thread"
    );

    // here you create your observer. I'll observe a string, so I need a Observer<String> obj
    Observer<String> stringObserver = new StringObserver();

    //now the observable will do his job on IO thread, but the result is emitted on mainThread
    observable
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(stringObserver);
}

class StringObserver extends DisposableObserver<String> {
    @Override
    public void onNext(String s) {
        //will be executed on main thread
    }

    @Override
    public void onError(Throwable e) {

        //will be executed on main thread
    }

    @Override
    public void onComplete() {

        //will be executed on main thread
    }
}