将AsyncTask转换为RxAndroid

Question

我有以下的方法，使用otto和AsyncTask将响应发布到UI。

private static void onGetLatestStoryCollectionSuccess(final StoryCollection storyCollection, final Bus bus) {
    new AsyncTask<Void, Void, Void>() {
        @Override
        protected Void doInBackground(Void... params) {
            bus.post(new LatestStoryCollectionResponse(storyCollection));
            return null;
        }
    }.execute();
}

我需要帮助将这个AsyncTask转换为使用RxAndroid库的RxJava。

- Fshamri

Answer 1

不要使用 .create()，而要使用 .defer()

Observable<File> observable = Observable.defer(new Func0<Observable<File>>() {
  @Override public Observable<File> call() {

    File file = downloadFile();

    return Observable.just(file);
  }
});

要了解更多细节，请查看https://speakerdeck.com/dlew/common-rxjava-mistakes

- Tarek360

---

这对我来说有点尴尬。它创建了一个输出已下载文件的Observable，通过延迟创建在后台执行。我认为使用Observable.fromCallable会更合适。 - Zackline

Answer 2

这是使用RxJava进行文件下载任务的示例

Observable<File> downloadFileObservable() {
    return Observable.create(new OnSubscribeFunc<File>() {
        @Override
        public Subscription onSubscribe(Observer<? super File> fileObserver) {
            try {
                byte[] fileContent = downloadFile();
                File file = writeToFile(fileContent);
                fileObserver.onNext(file);
                fileObserver.onCompleted();
            } catch (Exception e) {
                fileObserver.onError(e);
            }
            return Subscriptions.empty();
        }
    });
}

使用方法：

downloadFileObservable()
  .subscribeOn(Schedulers.newThread())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(observer); // you can post your event to Otto here

这将在新的线程上下载文件并在主线程上通知您。 OnSubscribeFunc已被弃用。更新代码以使用OnSubscribe。有关更多信息，请参见Github上的issue 802。 代码来自这里。

- LordRaydenMK

---

我无法使用这段代码，因为出现了错误："error: cannot find symbol class OnSubscribeFunc"。在我的gradle中，我已经添加了compile 'io.reactivex:rxandroid:1.0.1'和compile 'io.reactivex:rxjava:1.0.14'。 - Jachumbelechao Unto Mantekilla

@JachumbelechaoUntoMantekilla OnSubscribeFunc 已被弃用。您可以使用 OnSubscribe 替代。我已经更新了我的答案并链接到相关的 Github 问题。 - LordRaydenMK

取消订阅怎么样？如果当前的活动没有很快完成，这是必须的吧？ - shijin

onSubscribe已被弃用 - Hamid Reza

Answer 3

在您的情况下，您可以使用fromCallable。代码更少且自动发出onError。

Observable<File> observable = Observable.fromCallable(new Callable<File>() {
        @Override
        public File call() throws Exception {
            File file = downloadFile();
            return file;
        }
    });

使用lambda表达式：

Observable<File> observable = Observable.fromCallable(() -> downloadFile());

- Maksim Ostrovidov