RxJava前奏之原理分析

青旅半醒 2022-07-28 05:25 149阅读 0赞

## RxJava 之前奏：原理分析 ##

> 前言：RxJava在国内越来越火，本文只是原理分析的总结。代码不多，但有些绕，一步一步的推进，对于理解RxJava来说还是十分有好处的。但确实需要很大的毅力去坚持下去看完。

首先我们进入一个例子，关于猫的例子。

> 我们有个 Web API，能根据给定的查询请求搜索到整个互联网上猫的图片。每个图片包含可爱指数的参数（描述图片可爱度的整型值）。我们的任务将会下载到一个猫列表的集合，选择最可爱的那个，然后把它保存到本地。

首先定义实体类

public class Cat implements Comparable<Cat> { 
        /** * 图片 */
        String image;
    
        /** * 喜爱度 */
        int cuteness;
    
        @Override
        public int compareTo(Cat another) {
            return Integer.compare(cuteness, another.cuteness);
        }
    
    }

定义一个Uri类，用来定义图片保存的地址

public class Uri { 
    
    }

定义Api接口类，主要查询猫和保存猫的方法。

public interface Api { 
    
        /** * 查询所有猫 * @param query * @return */
        List<Cat> queryCats(String query);
    
        /** * 保存 * @param cat * @return */
        Uri store(Cat cat);
    
    }

定义业务类CastsHelp

public class CastsHelp { 
        Api api;
    
        //保存猫
        public Uri saveTheCutestCat(String query) {
            List<Cat> queryCats = api.queryCats(query);
            Cat cat = findCustest(queryCats);
    
            Uri savedUri = api.store(cat);
            return savedUri;
        }
    
        //查询最受喜爱的猫
        private Cat findCustest(List<Cat> queryCats) {
            return Collections.max(queryCats);
        }
    }

ok,我们的第一个例子写完了，让我们回过头来再仔细看一下这个例子，该业务逻辑类CastsHelp，里面包含了两个方法，其中saveTheCutestCat方法里面，包含了三个组合的方法，最终获取到保存猫的地址，并返回。这种组合方式即简单，又直白。一眼看去，清晰，明了。但我们往下看。

### 异步执行 ###

我们知道，对于一些网络请求，我们通常都需要进行异步去执行，那么，对于刚才说saveTheCutestCat方法，我们就不能如此写，对于api的中的方法，我们需要利用接口回调的方式回传结果。于是我们修改我们的代码。

public interface Api { 
    
    
        interface CatsQueryCallback{
            void onCatListReceived(List<Cat> cats);
            void onCatsQueryError(Exception e);
        }
    
    
        interface StoreCallback{
            void onCateStored(Uri uri);
            void onStoteFailed(Exception e);
        }
        /** * 查询所有猫 * @param query * @return */
        List<Cat> queryCats(String query,CatsQueryCallback castQueryCallback);
    
    
        /** * 保存 * @param cat * @return */
        Uri store(Cat cat,StoreCallback storeCallback);
    
    }

修改了一下api中的方法，因为对于猫和保存猫，分别为网络请求，和I/O操作，我们不能进行同步操纵，所以添加了两个接口，分别查询猫的和保存猫地址的结果的回调。

同时我们需要修改一下我们的业务逻辑类CastsHelp，也需要添加接口回调。

public interface CutestCatCallback{ 
            void onCutestCatSaved(Uri uri);
    
            void onCutestCatFail(Exception e);
        }
    
        Api api;
    
        public void saveTheCutestCat(String query,final CutestCatCallback custestCatCallback) {
            List<Cat> queryCats = api.queryCats(query,new CatsQueryCallback() {
    
                @Override
                public void onCatsQueryError(Exception e) {
                    //查询猫失败
                    custestCatCallback.onCutestCatFail(e);
                }
    
    
                @Override
                public void onCatListReceived(List<Cat> cats) {
                    Cat cat = findCustest(cats);
    
                    api.store(cat, new StoreCallback() {
    
                        @Override
                        public void onStoteFailed(Exception e) {
                            //保存猫失败
                            custestCatCallback.onCutestCatFail(e);
    
                        }
    
                        @Override
                        public void onCateStored(Uri uri) {
                            custestCatCallback.onCutestCatSaved(uri);
                        }
                    });
                }
            });
    
        }
    
        private Cat findCustest(List<Cat> queryCats) {
            return Collections.max(queryCats);
        }

代码开始复杂了，我们再来分析一下代码，在CastsHelp的saveTheCutestCat方法，我们传入了一个url，以及结果的回调，及该方法调用者用来实现的接口回调。  
\- 首先，调用api的queryCats方法查询所有的猫，因为该方法是异步的，所以我们传入一个接口的实现类，该接口包含了两个回调方法，onCatsQueryError（）查询失败的方法，在此方法中，我们直接调用调用者实现的接口告诉调用者获取信息出现异常，onCatListReceived（List cats）方法，表明查询猫成功，并返回猫列表。  
\- 我们在onCatListReceived（List cats）方法内，此时查询猫列表的数据已经返回，我们继续调用了获取最喜爱的猫方法findCustest，然后保存猫。store（）方法又包含了两个回调方法，在onStoteFailed（）方法中，依然如上。在成功的方法onCateStored（）中，我们将获取到的值通过调用者的实现接口进行回传给调用者。

好了，我们完成了异步任务的处理，代码复杂度大大的提升了，而且对于这么多接口回调，看着都心理难受。

### 接口回调的封装 ###

我们加入异步任务之后，整个程序在逻辑上的复杂度，代码的复杂度上都大大的提高了，但我们发现该接口有些共同点，我们可以进行封装。  
首先，先把之前添加的接口放在一起类比一下：

//业务逻辑类
        public interface CutestCatCallback{ 
            void onCutestCatSaved(Uri uri);
    
            void onCutestCatFail(Exception e);
        }
    
    
        //查询猫
        interface CatsQueryCallback{
            void onCatListReceived(List<Cat> cats);
            void onCatsQueryError(Exception e);
        }
    
        //保存猫
        interface StoreCallback{
            void onCateStored(Uri uri);
            void onStoteFailed(Exception e);
        }

看到三个接口，我们会发现，他们都有一个成功的回调和一个失败的回调，失败的回调参数一样，而成功的回调接口不一样。那么我们来抽取一下，抽取了一个如下的新接口回调：

public interface Callback<T> { 
    
        void onResult(T result);
    
        void onError(Exception e);
    
    }

新的接口回调，比较简单，成功和失败的方法，添加了泛型，那么我们把它移入项目中，这样关于我们的api接口又有所变动：

public interface Api { 
        /** * 查询所有猫 * @param query * @return */
        void queryCats(String query,Callback castQueryCallback);
    
    
        /** * 保存 * @param cat * @return */
        void store(Cat cat,Callback storeCallback);
    
    }

在看一下我们的业务逻辑类

public class CastsHelp { 
    
        Api api;
    
        public void saveTheCutestCat(String query,final Callback<Uri> custestCatCallback) {
            api.queryCats(query, new Callback<List<Cat>>() {
    
                @Override
                public void onResult(List<Cat> result) {
                    Cat cat = findCustest(result);
                    api.store(cat, new Callback<Uri>() {
    
                        @Override
                        public void onResult(Uri result) {
                            custestCatCallback.onResult(result);
                        }
    
                        @Override
                        public void onError(Exception e) {
                            custestCatCallback.onError(e);
                        }
                    });
                }
    
                @Override
                public void onError(Exception e) {
                    custestCatCallback.onError(e);
    
                }
            });
    
        }
    
        private Cat findCustest(List<Cat> queryCats) {
            return Collections.max(queryCats);
        }

逻辑上本有太大的变动，但在代码上，我们将之前定义的三个接口，统一改成了我们新定义的Callback接口。代码简洁了一些，同时也便于理解了一些。

### 再次进阶 ###

在上面我们的Demo中，我们通过定义一个泛型回调，大大降低了代码量，但还是存在一个问题，及我们每次调用这个方法，都需要传入两个参数，分别为url和接口回调。如果，我们能不能只请求一个参数url，同时返回一个对象，该对象包含了我们所需要的回调信息。

那么我们在思考一下，我们需要的是一个对象，该对象包含了一个只有接口回调为参数的方法，我们称之为AsyncJob。

public abstract class AsyncJob<T> { 
    
        public abstract void start(Callback<T> callback);
    
    }

我们让查询的方法等的返回值为该对象，则返回的该对象中调用start方法即可开始执行回调。

因为我们的Api中的两个方法是通过接口回掉来进行的，所以我们需要对Api类进行包装。ApiWrapper.

/** * 接口的包装类 备注人： Alex_MaHao * * @date 创建时间：2016年2月26日 上午10:11:28 */
    public class ApiWrapper { 
        Api api;
    
        public AsyncJob<List<Cat>> queryCats(final String query) {
            return new AsyncJob<List<Cat>>() {
    
                @Override
                public void start(final Callback<List<Cat>> callback) {
                    api.queryCats(query, new Callback<List<Cat>>() {
    
                        @Override
                        public void onResult(List<Cat> result) {
                            callback.onResult(result);
    
                        }
    
                        @Override
                        public void onError(Exception e) {
                            callback.onError(e);
    
                        }
                    });
                }
            };
    
        }
    
    
        public AsyncJob<Uri> store(final Cat cat) {
            return new AsyncJob<Uri>() {
    
                @Override
                public void start(final Callback<Uri> callback) {
                    api.store(cat, new Callback<Uri>() {
    
                        @Override
                        public void onResult(Uri result) {
                            callback.onResult(result);
                        }
    
                        @Override
                        public void onError(Exception e) {
                            callback.onError(e);
                        }
                    });
                }
    
            };
        }
    
    }

我们看一下我们的ApiWrapper中的方法，因为两个方法比较类似，我们只看queryCats方法，理解一下即可。如果单纯的看逻辑，会发现比较绕，我们看一下第一个方法应该怎么使用。

//调用方法，返回包含我们回调信息的asyncJob类
            AsyncJob<List<Cat>> async = new ApiWrapper().queryCats("query");
    
            //启动回调方法，获取回调结果
            async.start(new Callback<List<Cat>>() {
    
                @Override
                public void onResult(List<Cat> result) {
                    //所有猫的集合
                }
    
                @Override
                public void onError(Exception e) {
                    //错误回调
                }
            });

可以看到，我们的queryCats()传入一个url地址，返回一个包含我们回调接口的对象，这时候，我们只需要调用start()方法，即可获取到我们的回调结果。

那么，同时我们也需要修改一下CastsHelp中的代码

public class CastsHelp { 
    
        ApiWrapper apiWrapper;
    
        public AsyncJob<Uri> saveTheCutestCat(final String query) {
            return new AsyncJob<Uri>() {
    
                @Override
                public void start(final Callback<Uri> callback) {
                    apiWrapper.queryCats(query).start(new Callback<List<Cat>>() {
    
                        @Override
                        public void onResult(List<Cat> result) {
                            Cat cat = findCustest(result);
    
                            apiWrapper.store(cat).start(new Callback<Uri>() {
    
                                @Override
                                public void onResult(Uri result) {
                                    callback.onResult(result);
                                }
    
                                @Override
                                public void onError(Exception e) {
                                    callback.onError(e);
                                }
                            });
                        }
    
                        @Override
                        public void onError(Exception e) {
                            callback.onError(e);
                        }
                    });;
                }
            };
    
        }
    
        private Cat findCustest(List<Cat> queryCats) {
            return Collections.max(queryCats);
        }
    }

该方法也只是把我们的方法的返回值对象做了一下转换，修改了其返回值对象为我们的AsyncJob。

我们把该方法在进行整理，于是就有了如下代码

public class CastsHelp { 
    
        ApiWrapper apiWrapper;
    
        public AsyncJob<Uri> saveTheCutestCat(final String query) {
    
            //猫列表数据的接口回调
            final AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper
                    .queryCats(query);
    
            //查询最喜爱的猫
            final AsyncJob<Cat> cutestCatAsyncJob = new AsyncJob<Cat>() {
    
                @Override
                public void start(final Callback<Cat> callback) {
                    catsListAsyncJob.start(new Callback<List<Cat>>() {
    
                        @Override
                        public void onResult(List<Cat> result) {
                            callback.onResult(findCustest(result));
    
                        }
    
                        @Override
                        public void onError(Exception e) {
                            callback.onError(e);
                        }
                    });
                }
            };
    
            //获取保存的url地址的回掉对象并返回
            AsyncJob<Uri> storedUriAsyncJob = new AsyncJob<Uri>() {
    
                @Override
                public void start(final Callback<Uri> callback) {
                    cutestCatAsyncJob.start(new Callback<Cat>() {
    
                        @Override
                        public void onResult(Cat result) {
                            apiWrapper.store(result).start(callback);
                        }
    
                        @Override
                        public void onError(Exception e) {
                            callback.onError(e);
                        }
                    });
                }
            };
    
            return storedUriAsyncJob;
    
        }
    
        private Cat findCustest(List<Cat> queryCats) {
            return Collections.max(queryCats);
        }
    }

### map和flatMap ###

在我们的CastsHelp的saveTheCutestCat()方法中，有一个查询喜爱都最高的猫的列表，那么我们能不能这样理解呢，从输入和输出来看，我们期望将一个AsyncJob

public interface Func<T,R> { 
        R call(T t);
    }

该接口很简单，传入T，通过call方法返回R。

我们在AsyncJob中添加map方法，先看代码

public <R> AsyncJob<R> map(final Func<T,R> func){
            final AsyncJob<T> source = this;
    
            return new AsyncJob<R>(){
    
                @Override
                public void start(final Callback<R> callback) {
                    source.start(new Callback<T>() {
    
                        @Override
                        public void onResult(T result) {
                            //调用者实现的call方法
                            R r = func.call(result);
    
                            callback.onResult(r);
                        }
    
                        @Override
                        public void onError(Exception e) {
                            callback.onError(e);
                        }
                    });
                }
    
            };
    
        }

我们看一下这个方法，在之前我们这样分析，我们需要的是将AsyncJob

public  AsyncJob<Cat> map(final Func<List<Cat>,Cat> func){
    }

ok,我们在深入的看一下map方法，可以看到我们返回的对象，无非就是将当前的AsyncJob对象的回调信息与我们返回的信息所绑定，同时在onResult方法中将数据类型进行转换。

那么，在CastsHelp中的查询最喜爱的猫的方法，可改成如下所示：

final AsyncJob<Cat> cutestCatAsyncJob = catsListAsyncJob.map(new Func<List<Cat>, Cat>() {
    
                @Override
                public Cat call(List<Cat> t) {
                    return findCustest(t);
                }
            });

map对象，在转换的时候是将一个对象转换为另一个对象，通过的是返回值的方式，业绩是同步的方式进行的，那么我们会不会有另一种情况呢，即我们的call方法返回的也是一个回调信息，及也是异步的呢。

这种情况也是存在的，我们看一下我们的保存猫的方法，该方法是一个异步的任务，他不能直接返回值，而是返回一个回调的信息，我们看一下保存猫的方法，`public AsyncJob<Uri> store(final Cat cat)`，很明显，他不能向我们之前查询喜爱都最高的猫时直接获取Cat对象，并调用onResult方法并返回，他是一个回调对象，我们需要做的是将返回的回调对象和我们转换过后的相绑定。这时候我们定义了flatMap（）；

public <R> AsyncJob<R> flatMap(final Func<T, AsyncJob<R>> func ){
            final AsyncJob<T> source = this;
            return new AsyncJob<R>(){
    
                @Override
                public void start(final Callback<R> callback) {
                    source.start(new Callback<T>() {
    
                        @Override
                        public void onResult(T result) {
                            AsyncJob<R> mapped = func.call(result);
    
                            mapped.start(callback);
                        }
    
                        @Override
                        public void onError(Exception e) {
                            callback.onError(e);
    
                        }
                    });
    
                }
    
            };
        }

该方法将返回的Func回调对象与我们的flatMap对象绑定。那么，保存猫的方式也可以修改如下：

AsyncJob<Uri> storedUriAsyncJob = cutestCatAsyncJob.flatMap(new Func<Cat, AsyncJob<Uri>>() {
    
                @Override
                public AsyncJob<Uri> call(Cat t) {
                    // TODO Auto-generated method stub
                    return apiWrapper.store(t);
                }
            });

最后，贴上CastHelps的完整代码，

public AsyncJob<Uri> saveTheCutestCat(final String query) {
    
            //查询猫
            final AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper
                    .queryCats(query);
    
            //获取最喜爱的猫
            final AsyncJob<Cat> cutestCatAsyncJob = catsListAsyncJob.map(new Func<List<Cat>, Cat>() {
    
                @Override
                public Cat call(List<Cat> t) {
                    return findCustest(t);
                }
            });
    
            //获取地址
            AsyncJob<Uri> storedUriAsyncJob = cutestCatAsyncJob.flatMap(new Func<Cat, AsyncJob<Uri>>() {
    
                @Override
                public AsyncJob<Uri> call(Cat t) {
                    return apiWrapper.store(t);
                }
            });
    
            return storedUriAsyncJob;
    
        }

*  map和flatMap的总结  
    
    
     *  相同点：都是通过Func对象的Call方法，让调用者来实现不同数据的对象转换
     *  不同点：Call方法转换的目标结果不同，导致对结果的处理不同。如果是同步的转换，不涉及接口回掉，异步等，map处理。如果Call方法返回的是一个回调对象，那么我们需要去绑定，所以使用flatMap.

细心的可以看到我们saveTheCutestCat方法还可以精简

public AsyncJob<Uri> saveTheCutestCat(final String query) {
    
            return apiWrapper
                    .queryCats(query)
                    .map(new Func<List<Cat>, Cat>() {
    
                        @Override
                        public Cat call(List<Cat> t) {
    
    
                            return findCustest(t);
                        }
                    }).flatMap(new Func<Cat, AsyncJob<Uri>>() {
    
                        @Override
                        public AsyncJob<Uri> call(Cat t) {
    
                            return apiWrapper.store(t);
                        }
                    });
    
        }

### RxJava 的引入 ###

在引入RxJava 之前，我们之前写了那么多的代码势必要使用一下。那么怎么使用呢：

public static void main(String[] args) {
            CastsHelp castsHelp = new CastsHelp();
            castsHelp.saveTheCutestCat("ss").start(new Callback<Uri>() {
    
                @Override
                public void onResult(Uri result) {
                    //Uri 地址对象
                }
    
                @Override
                public void onError(Exception e) {
                    //异常
                }
            });
        }

ok,开始和RxJava进行类比。

AsyncJob类比Observable  
CallBack类比Observer  
start()类比 subscribe();

那么我们开始对我的类进行修改,在这之前需要导入rxjava的jar包

public class ApiWrapper { 
        Api api;
    
        public Observable<List<Cat>> queryCats(final String query) {
    
            return Observable.create(new OnSubscribe<List<Cat>>() {
    
                @Override
                public void call(final Subscriber<? super List<Cat>> arg0) {
                    api.queryCats(query, new CatsQueryCallback() {
    
                        @Override
                        public void onCatsQueryError(Exception e) {
                            arg0.onError(e);
                        }
    
                        @Override
                        public void onCatListReceived(List<Cat> cats) {
                            arg0.onNext(cats);
                        }
                    });
                }
            });
        }
    
        public Observable<Uri> store(final Cat cat) {
            return Observable.create(new OnSubscribe<Uri>() {
    
                @Override
                public void call(final Subscriber<? super Uri> subscriber) {
                    api.store(cat, new StoreCallback() {
    
                        @Override
                        public void onStoteFailed(Exception e) {
                            subscriber.onError(e);
                        }
    
                        @Override
                        public void onCateStored(Uri uri) {
                            subscriber.onNext(uri);
                        }
                    });
                }
            });
        }
    
    }

public class CastsHelp { 
    
        ApiWrapper apiWrapper;
    
        public Observable<Uri> saveTheCutestCat(final String query) {
            return apiWrapper
                    .queryCats(query)
                    .map(new Func1<List<Cat>, Cat>() {
    
                            @Override
                            public Cat call(List<Cat> cats) {
    
                                return findCustest(cats);
                            }
                        })
                    .flatMap(new Func1<Cat, Observable<Uri>>() {
    
                            @Override
                            public Observable<Uri> call(Cat cat) {
    
                                return apiWrapper.store(cat);
                            }
                        });
    
        }
    
        private Cat findCustest(List<Cat> queryCats) {
            return Collections.max(queryCats);
        }
    }

public class Test { 
        public static void main(String[] args) {
            CastsHelp castsHelp = new CastsHelp();
            castsHelp.saveTheCutestCat("ss").subscribe(new Observer<Uri>() {
    
                @Override
                public void onCompleted() {
    
                }
    
                @Override
                public void onError(Throwable arg0) {
    
                }
    
                @Override
                public void onNext(Uri arg0) {
    
                }
            });
        }
    }

RxJava的原理已经理解和整理完毕。下面就开进入到了RxJava的整理与总结了。没多少代码，就不共享了。

长路漫漫啊。加油！！！

研究资料  
[NotRxJava懒人专用指南][NotRxJava]

[NotRxJava]: http://www.devtf.cn/?p=323