什么时候应该使用RxJava Observable和Android上的简单回调？

Question

我正在为我的应用程序开发网络.所以我决定尝试Square的Retrofit.我看到他们支持简单Callback

@GET("/user/{id}/photo")
void getUserPhoto(@Path("id") int id, Callback<Photo> cb);

和RxJava的 Observable

@GET("/user/{id}/photo")
Observable<Photo> getUserPhoto(@Path("id") int id);

乍一看两者看起来都非常相似,但是当它实现时它变得有趣......

虽然简单的回调实现看起来类似于:

api.getUserPhoto(photoId, new Callback<Photo>() {
    @Override
    public void onSuccess() {
    }
});

这非常简单明了.随着Observable它迅速变得冗长和复杂.

public Observable<Photo> getUserPhoto(final int photoId) {
    return Observable.create(new Observable.OnSubscribeFunc<Photo>() {
        @Override
        public Subscription onSubscribe(Observer<? super Photo> observer) {
            try {
                observer.onNext(api.getUserPhoto(photoId));
                observer.onCompleted();
            } catch (Exception e) {
                observer.onError(e);
            }

            return Subscriptions.empty();
        }
    }).subscribeOn(Schedulers.threadPoolForIO());
}

那不是它.你仍然需要做这样的事情:

Observable.from(photoIdArray)
        .mapMany(new Func1<String, Observable<Photo>>() {
            @Override
            public Observable<Photo> call(Integer s) {
                return getUserPhoto(s);
            }
        })
        .subscribeOn(Schedulers.threadPoolForIO())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<Photo>() {
            @Override
            public void call(Photo photo) {
                //save photo?
            }
        });

我在这里错过了什么吗？或者这是一个使用Observables 的错误案例？何时/应该更喜欢Observable简单的回调？

更新

使用改造比上面的例子简单得多,正如@Niels在他的回答或杰克沃顿的示例项目U2020中所示.但基本上问题保持不变 - 何时应该使用某种方式？

Answer 1

对于简单的网络内容,RxJava优于Callback的优势非常有限.简单的getUserPhoto示例:

RxJava:

api.getUserPhoto(photoId)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<Photo>() {
            @Override
            public void call(Photo photo) {
               // do some stuff with your photo 
            }
     });

打回来:

api.getUserPhoto(photoId, new Callback<Photo>() {
    @Override
    public void onSuccess(Photo photo, Response response) {
    }
});

RxJava变体并不比Callback变体好多少.现在,让我们忽略错误处理.我们来看一张照片清单:

RxJava:

api.getUserPhotos(userId)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<List<Photo>, Observable<Photo>>() {
    @Override
    public Observable<Photo> call(List<Photo> photos) {
         return Observable.from(photos);
    }
})
.filter(new Func1<Photo, Boolean>() {
    @Override
    public Boolean call(Photo photo) {
         return photo.isPNG();
    }
})
.subscribe(
    new Action1<Photo>() {
    @Override
        public void call(Photo photo) {
            list.add(photo)
        }
    });

打回来:

api.getUserPhotos(userId, new Callback<List<Photo>>() {
    @Override
    public void onSuccess(List<Photo> photos, Response response) {
        List<Photo> filteredPhotos = new ArrayList<Photo>();
        for(Photo photo: photos) {
            if(photo.isPNG()) {
                filteredList.add(photo);
            }
        }
    }
});

现在,RxJava变体仍然不小,虽然使用Lambdas它会更接近Callback变体.此外,如果您有权访问JSON提要,那么当您只显示PNG时检索所有照片会有点奇怪.只需调整Feed即可显示PNG.

第一个结论

当您加载一个准备好格式正确的简单JSON时,它不会使您的代码库变小.

现在,让我们让事情变得更有趣.假设您不仅要检索userPhoto,而且还有Instagram克隆,并且要检索2个JSON:1.getUserDetails()2.getUserPhotos()

您希望并行加载这两个JSON,并且当两个JSON都加载时,应该显示该页面.回调变体将变得有点困难:您必须创建2个回调,将数据存储在活动中,如果加载了所有数据,则显示页面:

打回来:

api.getUserDetails(userId, new Callback<UserDetails>() {
    @Override
    public void onSuccess(UserDetails details, Response response) {
        this.details = details;
        if(this.photos != null) {
            displayPage();
        }
    }
});

api.getUserPhotos(userId, new Callback<List<Photo>>() {
    @Override
    public void onSuccess(List<Photo> photos, Response response) {
        this.photos = photos;
        if(this.details != null) {
            displayPage();
        }
    }
});

RxJava:

private class Combined {
    UserDetails details;
    List<Photo> photos;
}


Observable.zip(api.getUserDetails(userId), api.getUserPhotos(userId), new Func2<UserDetails, List<Photo>, Combined>() {
            @Override
            public Combined call(UserDetails details, List<Photo> photos) {
                Combined r = new Combined();
                r.details = details;
                r.photos = photos;
                return r;
            }
        }).subscribe(new Action1<Combined>() {
            @Override
            public void call(Combined combined) {
            }
        });

我们到了某个地方!RxJava的代码现在与回调选项一样大.RxJava代码更健壮; 想想如果我们需要加载第三个JSON(比如最新的视频)会发生什么？RxJava只需要很小的调整,而Callback变量需要在多个位置进行调整(在每个回调中我们需要检查是否检索了所有数据).

另一个例子; 我们想要创建一个自动完成字段,它使用Retrofit加载数据.每当EditText具有TextChangedEvent时,我们都不想进行网络搜索.快速键入时,只有最后一个元素应该触发调用.在RxJava上我们可以使用debounce运算符:

inputObservable.debounce(1, TimeUnit.SECONDS).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                // use Retrofit to create autocompletedata
            }
        });

我不会创建Callback变体,但你会明白这是更多的工作.

结论:当数据作为流发送时,RxJava非常好.Retrofit Observable同时推送流上的所有元素.与回调相比,这本身并不特别有用.但是当流上推送多个元素且时间不同时,你需要做与时序相关的事情,RxJava使代码更易于维护.

Answer 2

Observable的东西已经在Retrofit中完成了,所以代码可能是这样的:

api.getUserPhoto(photoId)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<Photo>() {
         @Override
            public void call(Photo photo) {
                //save photo?
            }
     });

Answer 3

在getUserPhoto()的情况下,RxJava的优势并不大.但是,当你为用户获取所有照片时,让我们再举一个例子,但只有当图像是PNG时,你才能访问JSON以在服务器端进行过滤.

api.getUserPhotos(userId)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.flatMap(new Func1<List<Photo>, Observable<Photo>>() {
    @Override
    public Observable<Photo> call(List<Photo> photos) {
         return Observable.from(photos);
    }
})
.filter(new Func1<Photo, Boolean>() {
    @Override
    public Boolean call(Photo photo) {
         return photo.isPNG();
    }
})
.subscribe(
    new Action1<Photo>() {
    @Override
        public void call(Photo photo) {
            // on main thread; callback for each photo, add them to a list or something.
            list.add(photo)
        }
    }, 
    new Action1<Throwable>() {
    @Override
        public void call(Throwable throwable) {
            // on main thread; something went wrong
            System.out.println("Error! " + throwable);
        }
    }, 
    new Action0() {
        @Override
        public void call() {
            // on main thread; all photo's loaded, time to show the list or something.
        }
    });

现在,JSON返回Photo的列表.我们将它们平面映射到单个项目.通过这样做,我们将能够使用过滤方法忽略不是PNG的照片.之后,我们将订阅,并在所有行完成后为每张单独的照片,错误处理程序和回调获取回调.

TLDR 点在这里; 回调只返回成功和失败的回调; RxJava Observable允许你做地图,缩小,过滤等等.

Answer 4

使用rxjava,您可以使用更少的代码执行更多操作.

假设您想在应用中实现即时搜索.使用回调你担心取消订阅上一个请求并订阅新请求,自己处理方向更改...我认为这是很多代码而且过于冗长.

使用rxjava非常简单.

public class PhotoModel{
  BehaviorSubject<Observable<Photo>> subject = BehaviorSubject.create(...);

  public void setUserId(String id){
   subject.onNext(Api.getUserPhoto(photoId));
  }

  public Observable<Photo> subscribeToPhoto(){
    return Observable.switchOnNext(subject);
  }
}

如果你想实现即时搜索,你只需要监听TextChangeListener并调用 photoModel.setUserId(EditText.getText());

在片段或活动的onCreate方法您订阅的返回photoModel.subscribeToPhoto()可观察的,它返回一个可观察总是发出的最新观察,(要求)emited的项目.

AndroidObservable.bindFragment(this, photoModel.subscribeToPhoto())
                 .subscribe(new Action1<Photo>(Photo photo){
      //Here you always receive the response of the latest query to the server.
                  });

此外,例如,如果PhotoModel是Singleton,则无需担心方向更改,因为无论您何时订阅,BehaviorSubject都会发出最后一次服务器响应.

通过这行代码,我们实现了即时搜索并处理方向更改.您是否认为可以使用较少代码的回调实现此功能？我对此表示怀疑.