0

我正在尝试学习 RxJava2，并将我的 AsyncTasks 转换为 Observables。

我有以下要转换的代码。

if(asyncTask.getStatus() == AsyncTask.Status.RUNNING){
    asyncTask.cancel();
}
asyncTask = new CustomTask();  
asyncTask.execute(input);

我尝试使用 Disposables 重新创建以下内容。

Disposable currentTask;
PublishSubject input = PublishSubject.create();

对于每个输入

if(currentTask != null) currentTask.dispose();

currentTask = input
    .map(// Network calls
         // returns CustomObject)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> {
               // do work with result
               }, throwable -> Log.e(TAG, throwable.toString()));

但是，currentTask始终为null。为什么？这是错误的方法吗？

4

1 回答 1

0

你使用Disposable正确，但我只能假设你在某个地方搞砸了这个主题。rx 中的主题既可以是发布者也可以是订阅者……而且主题不一定要等到subscribe(...)开始发布项目。因此，我不建议将您AsyncTask的 s 替换为任何类型的Subject.

您可以获得类似的、更具确定性的行为，但这样做：

Observable<CustomObject> networkObservable =
            Observable.create(emitter ->
                    {
                        try {
                            CustomObject object = doNetworking();
                            emitter.onNext(object);
                            emitter.onComplete();
                        } catch (Exception e) {
                            emitter.onError(e);
                        }
                    }
            );

if(currentTask != null) currentTask.dispose();

currentTask = networkObservable.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        // this next subscribe is similar to AsyncTask.execute() as it starts the stream
        .subscribe(result -> {
            // do work with result
            }, throwable -> Log.e(TAG, throwable.toString()));

此外，考虑调查SerialDisposable，您不必进行那些 null/dispose 检查

SerialDisposable serialDisposable = new SerialDisposable();

原子：在此容器上设置下一个一次性物品并处理前一个（如果有）或如果容器已被处理，则下一个处理。

Disposable disposable = networkObservable.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(...);

serialDisposable.set(disposable); // this will call dispose on the previous task if it exists
于 2017-10-10T19:48:35.783 回答