Android中的RxJava异步任务

Question

我正在尝试在 Android 中使用 RxJava 实现异步任务。 我尝试了以下代码，但没有成功。它在 UI 线程上执行。我正在使用以下版本的 RxAndroid 0.24.0。

try {
    Observable.just(someMethodWhichThrowsException())
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(s -> onMergeComplete());
}
catch (IOException e) {
    e.printStackTrace();
}

但是，以下内容对我来说是异步的。

Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            try {
                someMethodWhichThrowsException();
            } catch (IOException e) {
                e.printStackTrace();
            }

            subscriber.onCompleted();
        }
    });
    observable.subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe();

我正在尝试理解以下内容：

1. 它们有什么区别？
2. 创建异步任务时的最佳做法是什么？

Answer 1

1. 它们之间有什么区别？

Observable.just(someMethodWhichThrowsException())
    .subscribeOn(Schedulers.newThread())

这等价于：

Object someResult = someMethodWhichThrowsException();
Observable.just(someResult)
    .subscribeOn(Schedulers.newThread())

如您所见，这首先进行同步方法调用，然后将其传递给Observable.just 以成为 Observable。

Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(Subscriber<? super String> subscriber) {
            ...
        }
    })
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe();

但是，此方法将运行call 块中的代码订阅。你已经告诉它你想订阅一个新线程（subscribeOn(Schedulers.newThread())），所以订阅发生在一个新线程上，并且在订阅上运行的代码（call 块）也在那个线程上运行。这与调用Observable.defer 的行为类似。

1. 创建异步任务时的最佳做法是什么？

嗯，这取决于你和你想要的行为。有时您希望异步代码立即开始运行（在这种情况下，您可能希望使用其中一个操作符来缓存它）。我肯定会考虑为此使用Async Utils 库。

其他时候，您会希望它仅在订阅时运行（这是此处示例中的行为） - 例如，如果有副作用，或者如果您不在乎它何时运行而只想使用内置插件从 UI 线程中获取一些东西。 Dan Lew mentions Observable.defer 非常方便在转换为 Rx 期间将旧代码从 UI 线程中取出。

Answer 2

使用来自RxJava Async Utils 库的 Async.start()。这将调用您在另一个线程上提供的函数。

例子：

Observable<String> observable = Async.start(new Func0<String>() {
    @Override
    public String call() {
        try {
            return someMethodWhichThrowsException();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
});

正如您所注意到的，已检查的异常必须包装到 RuntimeExceptions 中。

另见https://github.com/ReactiveX/RxJava/wiki/Async-Operators#start