RxJava与响应式编程-蒲公英云

异步编程中主要关心的是线程间通信问题，Java中我们常用的一般有三种方式：

Synchronized/Lock ：通过线程阻塞的方式等待结果返回，代码写起来比较直观
Callbacks：异步方法没有返回值，但需要额外的回调参数（lambda或匿名类），在结果可用时调用它们。
Futures：异步方法立即返回Future 。异步线程计算任务结果，Future对象包装对它的访问。该值不会立即可用，并且可以轮询对象，直到该值可用。例如，运行Callable 任务的ExecutorService使用Future对象。

以上第一种需要通过阻塞线程实现同步，资源无法充分利用，不适合频繁的异步任务处理；第二种callback的方式，当出现多个连续异步调用时难以组合在一起，会代码难以阅读以及难以维护（称为“Callback Hell”）；第三种Futures避免了callback缩进的问题，能够以顺序的结构组织代码，但调用get仍然会阻塞代码，1.8引入了CompletableFuture提供了when、then等方法防止阻塞，但是使用体验仍然差强人意。

由于以上三种方式或多或少都存在不足，响应式编程作为新的异步通信方式逐渐流行起来。

响应式编程是一种关注于数据流（data streams）和变化传递（propagation of change）的异步编程方式。它可以用既有的编程语言表达静态（如数组）或动态（如事件源）的数据流。

RxJava是Java中最常用的响应式编程框架。RxJava的声明式API较好地解决了Callback带来的缩进问题；基于函数式编程思想的各种操作符可以让代码避免同步get这种阻塞式的逻辑。Java8的StreamApi也是一种响应式编程框架具备声明式的Api，但是它只能处理Cold流，不能像RxJava那样处理Hot流，也没有线程切换的能力，适用范围比较窄。

接下来我们通过实例来看一下RxJava相对于其他几种异步通信方式的优势：

RxJava VS Callbacks

先通过一个例子于Callback方式做一个对比：

在用户的UI上展示用户喜欢的top 5个商品的详细信息，如果不存在的话则调用推荐服务获取5个。实现此功能的实现三个接口：

获取用户喜欢的商品的ID的接口（userService.getFavorites）
获取商品详情信息接口（favoriteService.getDetails）
推荐商品与商品详情的服务（suggestionService.getSuggestions）

基于callback模式实现上面功能代码如下：

userService.getFavorites(userId, new Callback<List<String>>() { // ①
    public void onSuccess(List<String> list) { 
        if (list.isEmpty()) { 
            suggestionService.getSuggestions(new Callback<List<Favorite>>() {
                public void onSuccess(List<Favorite> list) {  // ③
                    UiUtils.submitOnUiThread(() -> { 
                        list.stream()
                            .limit(5)
                            .forEach(uiList::show); 
                    });
                }
                public void onError(Throwable error) { 
                    UiUtils.errorPopup(error);
                }
            });
        } else { // ②
            list.stream() 
                .limit(5)
                .forEach(favId -> favoriteService.getDetails(favId, new Callback<Favorite>() {
                    public void onSuccess(Favorite details) {
                        UiUtils.submitOnUiThread(() -> uiList.show(details));
                    }
                    public void onError(Throwable error) {
                        UiUtils.errorPopup(error);
                    }
                }
            ));
        }
    }
    public void onError(Throwable error) {
        UiUtils.errorPopup(error);
    }
});

三个接口请求是基于callback完成，如果结果正常调用onSuccess，结果异常则调用onError。
①：调用userService.getFavorites获取用户userId的推荐商品id列表，获取失败在UI上显示错误
② ：获取id列表不为空时，则取前5个数据循环调用favoriteService.getDetails获取商品详情并在UI上显示；若id列表为空则调用suggestionService.getSuggestion获取推荐商品及详情
③：获取推荐商品后，通过stream操作符的limit取5个元素显示到UI上

上述使用了大量callback，这些代码逻辑晦涩难懂，且存在重复代码，下面我们使用RxJava配合Retrofit实现同样的需求：

userService.getFavorites(userId)
            .flatMap(Observable::fromIterable)
            .take(5)
            .flatMap(favoriteService::getDetails) 
             //.storted()
            .switchIfEmpty(suggestionService.getSuggestions()) 
            .take(5)
            .observeOn(AndroidSchedules.MainThread) 
            .subscribe(uiList::show, UiUtils::errorPopup);

getFavorites获取userId对应的商品列表Observable>，通过fromIterable将list转换为Item的Stream，通过take获取前5个
接下来通过flatMap把每个商品id转换为详情信息（favoriteService::getDetails）。
如果getFavorites返回空列表switchIfEmpty命中，然后调用suggestionService.getSuggestions()服务获取推荐的商品详情，
subscribeOn把当前线程切换到UI调度器来执行，通过subscribe方法激活整个流处理链，然后在UI线程上绘制商品详情列表或者显示错误。
由于getDetails是一个异步请求，不能保证返回的流顺序固定，如果有需要可以通过storted操作符进行排序

如上，RxJava的代码基于声明式编程，比较通俗易懂，代码量也比较少，不含有重复的代码。

RxJava VS Futures

在通过一个例子对比一下RxJava与Future的区别：

首先我们获取一个id列表，然后根据id分别获取对应的name和统计数据，然后组合每个id对应的name和统计数据为一个新的数据，最后输出所有组合对的值。使用CompletableFuture来实现这个功能，以便保证整个过程是异步的，并且每个id对应的处理是并发的：

CompletableFuture<List<String>> ids = ifhIds(); // ①
CompletableFuture<List<String>> result = ids.thenComposeAsync(l -> { // ②
        Stream<CompletableFuture<String>> zip =
            l.stream().map(i -> { // ③
                    CompletableFuture<String> nameTask = ifhName(i); 
                    CompletableFuture<Integer> statTask = ifhStat(i); 
                    return nameTask.thenCombineAsync(statTask, (name, stat) -> 
                        "Name " + name + " has stats " + stat); 
            });
        List<CompletableFuture<String>> combinationList =  
            zip.collect(Collectors.toList());  // ④
        CompletableFuture<String>[] combinationArray = 
            combinationList.toArray(new CompletableFuture[combinationList.size()]); // ⑤
        CompletableFuture<Void> allDone = CompletableFuture.allOf(combinationArray); // ⑥
        return allDone.thenApply(v ->
             combinationList.stream()                                                                                                  
                .map(CompletableFuture::join)                                                                                                      
                .collect(Collectors.toList())); // ⑦
});
List<String> results = result.join();  // ⑧

①：调用ifhIds方法异步返回了一个CompletableFuture对象，其内部保存了id列表
②：调用ids的thenComposeAsync方法返回一个新的CompletableFuture对象，新CompletableFuture对象是lambda执行结果
③：lambda内获取id列表的流对象，使用map通过id分别获取为name与统计信息进行合并。
④：把流中元素收集保存到了combinationList列表里面。
⑤：把列表转换为了数组，这是因为后面的allOf操作符的参数必须为数组。
⑥：把combinationList列表中的所有CompletableFuture对象转换为了一个allDone（等所有CompletableFuture对象的任务执行完毕），到这里我们调用allDone的get（）方法就可以等待所有异步处理执行完毕，但是我们目的是想获取到所有异步任务的执行结果，所以代码7在allDone上施加了thenApply运算，意在等所有任务处理完毕后调用所有CompletableFuture的join方法获取每个任务的执行结果，然后收集为列表后返回一个新的CompletableFuture对象。
⑧：在新的CompletableFuture上调用join方法获取所有执行结果列表。

用RxJava实现通用的需求：

Observable<String> ids = ifhrIds(); 
Observable<String> combinations = ids.flatMap(id -> { 
           return ifhrName(id).zipWith(ifhrStat(id),
                 (name, stat) -> "Name " + name + " has stats " + stat));
    });
Observable<List<String>> result = combinations.toList(); 
List<String> results = result.blockingFirst();