RxJava与响应式编程

柔情只为你懂 2023-07-03 03:00 5阅读 0赞

异步编程中主要关心的是线程间通信问题，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个。实现此功能的实现三个接口：

1.  获取用户喜欢的商品的ID的接口（userService.getFavorites）
2.  获取商品详情信息接口（favoriteService.getDetails）
3.  推荐商品与商品详情的服务（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<List<Int>>，通过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();

*  调用ifhIds方法异步返回了一个Observable对象，其内部保存了id列表
 *  ifhrName和ifhrState分别根据id返回Observable对象，然后通过zip进行合并
 *  toList将流式数据转成list，最后通过blockingFirst()同步获取结果。

如上代码使用RxJava相比使用CompletableFuture来说，更简洁，更通俗易懂。

## 总结 ##

RxJava作为一个响应式框架有以下特点

*  可编排性（Composability） 以及 可读性（Readability）
 *  丰富的操作符
 *  灵活的线程切换
 *  Cold和Hot之间灵活转换
 *  背压（backpressure） 即消费者能够反向告知生产者生产内容的速度的能力

RxJava等响应式框架不仅仅是一个异步通信框架，更重要的意义在于将传统的同步的命令式的开发范式引导为一个异步的声明式的开发范式，这将非常有利于我们编写出更加高性能的应用。