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

When should one use RxJava Observable and when simple Callback on Android?

提问by Martynas Jurkus

I'm working on networking for my app. So I decided to try out Square's Retrofit. I see that they support simple Callback

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

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

and RxJava's Observable

和 RxJava 的 Observable

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

Both look pretty similar at first glance, but when it gets to implementation it gets interesting...

乍一看，两者看起来非常相似，但是当它开始实施时，它变得有趣......

While with simple callback implementation would look similar to this:

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

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

which is quite simple and straightforward. And with Observableit quickly gets verbose and quite complicated.

这非常简单明了。并且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());
}

And that is not it. You still have to do something like this:

事实并非如此。你仍然需要做这样的事情：

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?
            }
        });

Am I missing something here? Or is this a wrong case to use Observables? When would/should one prefer Observableover simple Callback?

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

Update

更新

Using retrofit is much simpler than example above as @Niels showed in his answer or in Jake Wharton's example project U2020. But essentially the question stays the same - when should one use one way or the other?

正如@Niels 在他的回答或 Jake Wharton 的示例项目U2020 中所示，使用改造比上面的示例简单得多。但基本上问题保持不变 - 什么时候应该使用一种方式或另一种方式？

回答by Niels

For simple networking stuff, the advantages of RxJava over Callback is very limited. The simple getUserPhoto example:

对于简单的网络内容，RxJava 相对于回调的优势非常有限。简单的 getUserPhoto 示例：

RxJava:

RxJava：

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

Callback:

打回来：

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

The RxJava variant is not much better than the Callback variant. For now, let's ignore the error handling. Let's take a list of photos:

RxJava 变体并不比回调变体好多少。现在，让我们忽略错误处理。来一张照片清单吧：

RxJava:

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)
        }
    });

Callback:

打回来：

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);
            }
        }
    }
});

Now, the RxJava variant still isn't smaller, although with Lambdas it would be getter closer to the Callback variant. Furthermore, if you have access to the JSON feed, it would be kind of weird to retrieve all photos when you're only displaying the PNGs. Just adjust the feed to it only displays PNGs.

现在，RxJava 变体仍然不小，尽管使用 Lambdas 它将更接近回调变体。此外，如果您可以访问 JSON 提要，那么当您只显示 PNG 时检索所有照片会有点奇怪。只需将提要调整为仅显示 PNG。

First conclusion

第一个结论

It doesn't make your codebase smaller when you're loading a simple JSON that you prepared to be in the right format.

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

Now, let's make things a bit more interesting. Let's say you not only want to retrieve the userPhoto, but you have an Instagram-clone, and you want to retrieve 2 JSONs: 1. getUserDetails() 2. getUserPhotos()

现在，让我们让事情变得更有趣。假设您不仅想检索 userPhoto，而且您有一个 Instagram 克隆，并且您想检索 2 个 JSON： 1. getUserDetails() 2. getUserPhotos()

You want to load these two JSONs in parallel, and when both are loaded, the page should be displayed. The callback variant will become a bit more difficult: you have to create 2 callbacks, store the data in the activity, and if all the data is loaded, display the page:

您希望并行加载这两个 JSON，并且在加载这两个 JSON 时，应该会显示该页面。回调变体会变得有点困难：您必须创建 2 个回调，将数据存储在 Activity 中，如果所有数据都加载完毕，则显示页面：

Callback:

打回来：

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:

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) {
            }
        });

We are getting somewhere! The code of RxJava is now as big as the callback option. The RxJava code is more robust; Think of what would happen if we needed a third JSON to be loaded (like the latest Videos)? The RxJava would only need a tiny adjustment, while the Callback variant needs to be adjusted in multiple places (on each callback we need to check if all data is retrieved).

我们正在到达某个地方！RxJava 的代码现在和回调选项一样大。RxJava 代码更健壮；想想如果我们需要加载第三个 JSON（比如最新的视频）会发生什么？RxJava 只需要很小的调整，而 Callback 变体需要在多个地方进行调整（在每个回调中，我们需要检查是否检索到所有数据）。

Another example; we want to create an autocomplete field, which loads data using Retrofit. We don't want to do a webcall every time an EditText has a TextChangedEvent. When typing fast, only the last element should trigger the call. On RxJava we can use the debounce operator:

另一个例子; 我们想创建一个自动完成字段，它使用 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
            }
        });

I won't create the Callback variant but you will understand this is much more work.

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

Conclusion: RxJava is exceptionally good when data is sent as a stream. The Retrofit Observable pushes all elements on the stream at the same time. This isn't particularly useful in itself compared to Callback. But when there are multiple elements pushed on the stream and different times, and you need to do timing-related stuff, RxJava makes the code a lot more maintainable.

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

回答by Niels

The Observable stuff is already done in Retrofit, so the code could be this:

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

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

回答by Niels

In the case of getUserPhoto() the advantages for RxJava aren't great. But let's take another example when you'll get all the photos for a user, but only when the image is PNG, and you don't have access to the JSON to do the filtering on the serverside.

在 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.
        }
    });

Now the JSON returns a list of Photo's. We'll flatMap them to individual items. By doing so, we'll be able to use the filter method to ignore photos which are not PNG. After that, we'll subscribe, and get a callback for each individual photo, an errorHandler, and a callback when all rows have been completed.

现在 JSON 返回照片列表。我们会将它们 flatMap 到单个项目。通过这样做，我们将能够使用过滤器方法来忽略不是 PNG 的照片。之后，我们将订阅，并获得每张照片的回调、errorHandler 和所有行完成后的回调。

TLDRPoint here being; the callback only returns you a callback for succes and failure; the RxJava Observable allows you to do map, reduce, filter and many more stuff.

TLDR点在这里；回调只返回成功和失败的回调；RxJava Observable 允许你做映射、缩减、过滤等等。

回答by Roger Garzon Nieto

With rxjava you can do more things with less code.

使用 rxjava，您可以用更少的代码做更多的事情。

Let′s assume that you want to implement instant search in your app. With callbacks you have worried about unsubscribing the previous request and subscribe to the new one, handle orientation change yourself... I think it′s a lot of code and too verbose.

假设您想在您的应用中实现即时搜索。使用回调，您担心取消订阅先前的请求并订阅新的请求，自己处理方向更改...我认为它的代码很多而且过于冗长。

With rxjava is very simple.

用 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);
  }
}

if you want to implement instant search you only have to listen for TextChangeListener and call to photoModel.setUserId(EditText.getText());

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

In onCreate method of Fragment or activity you subscribe to the Observable that returns photoModel.subscribeToPhoto(), it returns an Observable that always emit the items emited by the latest Observable(request).

在您订阅返回 photoModel.subscribeToPhoto() 的 Observable 的 Fragment 或活动的 onCreate 方法中，它返回一个始终发出最新 Observable(request) 发出的项目的 Observable。

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

Also, if PhotoModel is a Singleton, for instance, you don't need to worry about orientation changes, because BehaviorSubject emits the last server response, regardless of when you subscribe.

此外，例如，如果 PhotoModel 是单例，则您无需担心方向更改，因为 BehaviorSubject 会发出最后一个服务器响应，而不管您何时订阅。

With this lines of code we have implemented an instant search and handle orientation changes. Do you think that you can implement this with callbacks with less code? I doubt it.

通过这几行代码，我们实现了即时搜索并处理方向变化。你认为你可以用更少的代码用回调来实现吗？我对此表示怀疑。

回答by Dzmitry Lazerka

We usually go with the following logic:

我们通常采用以下逻辑：

1. If it's a simple one-response call, then Callback or Future is better.
2. If it's a call with multiple responses (stream), or when there are complex interaction between different calls (see @Niels' answer), then Observables are better.

1. 如果是简单的单响应调用，则 Callback 或 Future 更好。
2. 如果是具有多个响应（流）的调用，或者不同调用之间存在复杂的交互（参见 @Niels 的回答），那么 Observables 会更好。

回答by Reza

I personally prefer to use Rx to get api responses in cases that I have to do filter, map or something like that on the data Or in cases that I have to do another api calls based on previous calls responses

我个人更喜欢在我必须对数据进行过滤、映射或类似操作的情况下使用 Rx 来获取 api 响应，或者在我必须根据以前的调用响应进行另一个 api 调用的情况下

回答by user_3380739

By the samples and conclusions in other answers, I think there is no big difference for simple one or two-steps tasks. However, Callback is simple and straightforward. RxJava is more complicated and too big for simple task. There is the third solution by: AbacusUtil. Let me implement above use cases with all three solutions: Callback, RxJava, CompletableFuture(AbacusUtil) with Retrolambda:

通过其他答案中的样本和结论，我认为简单的一两步任务没有太大区别。然而，回调是简单明了的。RxJava 更复杂，对于简单的任务来说太大了。有第三个解决方案：AbacusUtil。让我使用所有三个解决方案来实现上述用例：Callback、RxJava、CompletableFuture(AbacusUtil) 和Retrolambda：

Fetch photo from network and save/display on device:

从网络获取照片并保存/显示在设备上：

// By Callback
api.getUserPhoto(userId, new Callback<Photo>() {
    @Override
    public void onResponse(Call<Photo> call, Response<Photo> response) {
        save(response.body()); // or update view on UI thread.
    }

    @Override
    public void onFailure(Call<Photo> call, Throwable t) {
        // show error message on UI or do something else.
    }
});

// By RxJava
api.getUserPhoto2(userId) //
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(photo -> {
            save(photo); // or update view on UI thread.
        }, error -> {
            // show error message on UI or do something else.
        });

// By Thread pool executor and CompletableFuture.
TPExecutor.execute(() -> api.getUserPhoto(userId))
        .thenRunOnUI((photo, error) -> {
            if (error != null) {
                // show error message on UI or do something else.
            } else {
                save(photo); // or update view on UI thread.
            }
        });

Load user details and photo in parallel

并行加载用户详细信息和照片

// By Callback
// ignored because it's little complicated

// By RxJava
Observable.zip(api.getUserDetails2(userId), api.getUserPhoto2(userId), (details, photo) -> Pair.of(details, photo))
        .subscribe(p -> {
            // Do your task.
        });

// By Thread pool executor and CompletableFuture.
TPExecutor.execute(() -> api.getUserDetails(userId))
          .runOnUIAfterBoth(TPExecutor.execute(() -> api.getUserPhoto(userId)), p -> {
    // Do your task
});

回答by Samuel Gruetter

It looks like you're reinventing the wheel, what you're doing is already implemented in retrofit.

看起来您正在重新发明轮子，您正在做的事情已经在改造中实施了。

For an example, you could look at retrofit's RestAdapterTest.java, where they define an interfacewith Observable as return type, and then use it.

例如，您可以查看改造的RestAdapterTest.java，他们在其中定义了一个以 Observable 作为返回类型的接口，然后使用它。