在 Android 上什么时候应该使用 RxJava Observable，什么时候应该使用简单的 Callback？

我正在为我的应用程序开发网络。所以我决定尝试一下 Square 的Retrofit https://github.com/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?
            }
        });

我在这里错过了什么吗？或者这是一个错误的使用情况Observable是？ 人们什么时候会/应该更喜欢Observable通过简单的回调？

Update

使用改造比上面的例子简单得多，正如 @Niels 在他的答案或 Jake Wharton 的示例项目中所示U2020 https://github.com/JakeWharton/u2020。但本质上问题仍然是一样的——什么时候应该使用一种方式或另一种方式？

对于简单的网络事物，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 变体仍然不小，尽管使用 Lambda 后它会更接近 Callback 变体。 此外，如果您有权访问 JSON feed，那么当您只显示 PNG 时检索所有照片会有点奇怪。只需将 feed 调整为仅显示 PNG 即可。

第一个结论

当您加载准备采用正确格式的简单 JSON 时，它不会使您的代码库变小。

现在，让我们让事情变得更有趣一些。假设您不仅想要检索 userPhoto，而且您有一个 Instagram 克隆，并且想要检索 2 个 JSON： 1. 获取用户详细信息() 2. 获取用户照片()

您想要并行加载这两个 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
            }
        });

我不会创建回调变体，但您会明白这是更多的工作。

结论： 当数据作为流发送时，RxJava 表现得非常好。 Retrofit Observable 同时推送流上的所有元素。 与回调相比，这本身并不是特别有用。但是，当有多个元素在不同时间推送到流上时，并且您需要执行与计时相关的操作，RxJava 使代码更易于维护。