RxJava理解系列（一）

有人说RxJava非常好用，那么，它究竟好用在哪里？今天来具体分析下。
首先，先来阐述RxJava到底是什么，RxJava官方的解释是：“a library for composing asynchronous and event-based programs using observable sequences for the Java VM”，其核心就是“asynchronous”这个词，直白的说，RxJava就是一个实现异步操作的库。
那为什么大家会觉得RxJava好用，而不是使用AsyncTask/Handler...？这里可以归结一个词，简洁。
举个例子，我们要从网络上获取图片然后显示。利用AsyncTask的做法是这样的：

public class ImageTask extends AsyncTask<String, Void, Bitmap> {

        @Override
        protected Bitmap doInBackground(String... params) {

            String imageUrl = params[0];
            try {
                URL url = new URL(imageUrl);
                HttpURLConnection conn = (HttpURLConnection) url.openConnection();
                InputStream is = conn.getInputStream();
                Bitmap bitmap = BitmapFactory.decodeStream(is);
                if (bitmap != null) {
                    return bitmap;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }

        @Override
        protected void onPostExecute(Bitmap bitmap) {
            super.onPostExecute(bitmap);
            if(bitmap!=null){
                showBitMap();
            }
        }
    }

如果使用RxJava,那么实现方式是这样的：

   Observable.just(imageUrl)
                .map(new Func1<String, Bitmap>() {
                    @Override
                    public Bitmap call(String url) {
                        return getBitmapFromUrl(url);
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<Bitmap>() {
                    @Override
                    public void call(Bitmap bitmap) {
                        showBitMap();
                    }
                });

 可能有的朋友要说话了，（没看出什么区别啊，好用个毛线啊）。其实我们观察下就能发现，RxJava的这个实现，就是一条链式调用，在逻辑上没有任何的嵌套，简单明了，更进一步说，当我们需求变得复杂时，比如在大量的图片资源中选择一张，亦或者是选择前几张图片的时候，你还能在你那片代码中能迅速理清逻辑，并快速调整？RxJava在处理复杂逻辑时，一条链式调用，虽然很长，但胜在逻辑清晰。
 接下来，以一个例子具体阐述下RxJava到底逻辑清晰在哪里。有一个需求，谁是最可爱的人（sb需求，哈哈哈）。就是网络请求查询一组图片，每张图片有一个可爱系数（一个整型值），而我们的任务，就是下载一组可爱人的照片集合，然后选择最可爱的那个。

首先，定义一个简单的数据结构：

   public class CutestPeople implements Comparable<CutestPeople> {

    private Bitmap image;
    private int cuteness;

    @Override
    public int compareTo(CutestPeople cutestPeople) {
        return Integer.compare(cuteness, cutestPeople.cuteness);
    }
   }

然后，提供获取查询和存储的api:

 public interface QueryApi {
    /** * 获取集合 * @param query * @return */
    List<CutestPeople>  queryCuestPeople(String query);
    /** * 获取需要保存的Uri * @param people * @return */
    Uri store(CutestPeople people);
}

最后，整体逻辑大体是这个样子的：

public class DataHelper {
    private QueryApi mApi;

    public Uri saveTheCutestPeople(String query) {
        List<CutestPeople> cutestPeoples = mApi.queryCuestPeople(query);
        CutestPeople cutestPeople = findCutestPeople(cutestPeoples);
        Uri savedUrl = mApi.store(cutestPeople);
        return savedUrl;
    }

    /** * 获取 * @param people * @return */
    public CutestPeople findCutestPeople(List<CutestPeople> people) {
        return Collections.max(people);
    }
}

等等，似乎少了什么，没错，异常处理机制，赶紧加上：

   try {
       List<CutestPeople> cutestPeoples = mApi.queryCuestPeople(query);
       CutestPeople cutestPeople = findCutestPeople(cutestPeoples);
       Uri savedUrl = mApi.store(cutestPeople);
       return savedUrl;
   } catch (Exception e) {
       e.printStackTrace();
       return error;
   }

看看上面的代码，我们继续优化，加入异步回调方式，现在QueryApi变成这样：

public interface QueryApi {


    interface CuestPeopleQueryCallback {
        void onCuestPeopleReceived(List<CutestPeople> people);
        void onQueryError(Exception e);
    }

    interface StoreCallback {
        void onCuestPeopleStore(Uri uri);
        void onStoreError(Exception e);
    }

    /** * 获取可爱人儿集合 * * @param query * @return */
    List<CutestPeople> queryCutestPeople(String query,CutestPeopleQueryCallback callback);

    /** * 保存最可爱的人到本地 * * @param people * @return */
    Uri store(CutestPeople people,StoreCallback callback);
}

相应的，我们的逻辑处理也进行相应的更改：

public class DataHelper {

    private QueryApi mApi;

    public interface CutestPeopleCallback {
        void onCutestPeopleSaved(Uri uri);
        void onError(Exception e);
    }

    public void saveTheCuestPeople(String query, final CutestPeopleCallback cutestCallback) {
     mApi.queryCutestPeople(query, new QueryApi.CutestPeopleQueryCallback() {
     @Override
     public void onCutestPeopleReceived(List<CutestPeople> people) {
        CutestPeople cutestPeople = findCutestPeople(people);
        mApi.store(cutestPeople, new QueryApi.StoreCallback() {
        @Override
        public void onCutestPeopleStore(Uri uri) {
                             cutestPeopleCallback.onCutestPeopleSaved(uri);
        }

        @Override
        public void onStoreError(Exception e) {
            cutestPeopleCallback.onError(e);
            }
          });
        }

            @Override
            public void onQueryError(Exception e) {
                cutestPeopleCallback.onError(e);
            }
        });
    }

    /** * 获取 * * @param people * @return */
    public CutestPeople findCutestPeople(List<CutestPeople> people) {

        return Collections.max(people);
    }
}

这样的代码看起来怎样？是不是非常的冗余，对于每一个异步操作，我们都需要定义相应的回调接口手动的一条条的进行插入，而且缺少错误传递机制。
接下来，我们继续修改上面的代码，目前，回调接口是这样的：

void onCutestPeopleReceived(List<CutestPeople> people);
void onCutestPeopleStore(Uri uri);
...
void onQueryError(Exception e);
void onStoreError(Exception e);

我们不能修改api中的方法，但是我们可以将上面的回调接口用泛型进行包装：

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

接下来，我们将Api的方法进行封装：

public class ApiWarpper {

    private QueryApi mApi;

    public void queryQuestPeople(String query, final MyCallback<List<CutestPeople>> myCallback) {

        mApi.queryCutestPeople(query, new QueryApi.CutestPeopleQueryCallback() {
            @Override
            public void onCutestPeopleReceived(List<CutestPeople> people) {
                myCallback.onGetResult(people);
            }

            @Override
            public void onQueryError(Exception e) {

                myCallback.onError(e);
            }
        });

    }

    public void storeUri(CutestPeople people, final MyCallback<Uri> myCallback) {
        mApi.store(people, new QueryApi.StoreCallback() {
            @Override
            public void onCutestPeopleStore(Uri uri) {
                myCallback.onGetResult(uri);
            }

            @Override
            public void onStoreError(Exception e) {
                myCallback.onError(e);
            }
        });
    }
}

最后，我们的逻辑处理优化后如下：

public class DataHelper {

    ApiWrapper mApiWrapper;

    public void saveTheCutestPeople(String query, final MyCallback<Uri> myCallBack) {

        mApiWrapper.queryQuestPeople(query, new MyCallback<List<CutestPeople>>() {
            @Override
            public void onGetResult(List<CutestPeople> result) {

                CutestPeople cutestPeople = findCutestPeople(result);
                mApiWrapper.storeUri(cutestPeople, myCallBack);
            }

            @Override
            public void onError(Exception e) {
                myCallBack.onError(e);
            }
        });

    }

    /** * 获取 * * @param people * @return */
    public CutestPeople findCutestPeople(List<CutestPeople> people) {

        return Collections.max(people);
    }
}

现在来说，目前代码大体没有什么问题，但是，我们能不能将异步操作分解为更小的操作，即每个异步操作仅仅只携带一个参数对象，然后返回一些携带着回调信息的临时对象。
接下来，定义一个通用的临时对象，该对象只携带一个参数对象：

public abstract class MyAsyncTask<T> {
    public abstract void start(MyCallback<T> callback);
}

修改ApiWrapper中的内容，如下所示：

public class ApiWrapper {

    private QueryApi mApi;

    public MyAsyncTask<List<CutestPeople>> queryQuestPeople(final String query) {
    return new MyAsyncTask<List<CutestPeople>>() {
    @Override
    public void start(final MyCallback<List<CutestPeople>> callback) {
       mApi.queryCutestPeople(query, new QueryApi.CutestPeopleQueryCallback() {
    @Override
    public void onCutestPeopleReceived(List<CutestPeople> people) {
       callback.onGetResult(people);
    }

    @Override
    public void onQueryError(Exception e) {
       callback.onError(e);
            }
         });
        }
      };
   }

    public MyAsyncTask<Uri> storeUri(final CutestPeople people){
        return new MyAsyncTask<Uri>() {
            @Override
            public void start(final MyCallback<Uri> callback) {

                mApi.store(people, new QueryApi.StoreCallback() {
                    @Override
                    public void onCutestPeopleStore(Uri uri) {
                        callback.onGetResult(uri);
                    }

                    @Override
                    public void onStoreError(Exception e) {

                        callback.onError(e);
                    }
                });
            }
        };
    }
}

相应的，逻辑部分进行相应的调整：

public class DataHelper {

    ApiWrapper mApiWrapper;

    public MyAsyncTask<Uri> saveTheCuestPeople(final String query) {

        final MyAsyncTask<List<CutestPeople>> listMyAsyncTask = mApiWrapper.queryQuestPeople(query);

        final MyAsyncTask<CutestPeople> myCutestAsyncTask = new MyAsyncTask<CutestPeople>() {

            @Override
            public void start(final MyCallback<CutestPeople> callback) {

             @Override
             public void onGetResult(List<CutestPeople> result) {                       
             callback.onGetResult(findCutestPeople(result));
             }

             @Override
             public void onError(Exception e) {
             callback.onError(e);
                }
             });
            }
        };

     MyAsyncTask<Uri> uriMyAsyncTask = new MyAsyncTask<Uri>() {
          @Override
          public void start(final MyCallback<Uri> callback) {
           myCutestAsyncTask.start(new MyCallback<CutestPeople>() {
                    @Override
                    public void onGetResult(CutestPeople result) {
                        mApiWrapper.storeUri(result)
                                .start(new MyCallback<Uri>() {
                                 @Override
                                public void onGetResult(Uri result) {
                                        callback.onGetResult(result);
                                    }

                                  @Override
                                  public void onError(Exception e) {
                                        callback.onError(e);
                                    }
                                });
                    }

                    @Override
                    public void onError(Exception e) {
                        callback.onError(e);
                    }
                });
            }
        };
        return uriMyAsyncTask;
    }


    /** * 获取 * * @param people * @return */

    public CutestPeople findCutestPeople(List<CutestPeople> people) {
        return Collections.max(people);
    }
}

这样的修改似乎更加复杂，但是，我们能够返回一个组合的操作，这样我们的Activity或者Fragment就能通过组合的工作来进行操作。

将我们的代码简化表达，如下所示：


public class DataHelper {

    ApiWrapper mApiWrapper;

    public MyAsyncTask<Uri> saveTheCuestPeople(final String query) {

      final MyAsyncTask<List<CutestPeople>> listMyAsyncTask = mApiWrapper.queryQuestPeople(query);

      final MyAsyncTask<CutestPeople> myCutestAsyncTask = new MyAsyncTask<CutestPeople>() { ...};
      MyAsyncTask<Uri> uriMyAsyncTask = new MyAsyncTask<Uri>() {...};
      return uriMyAsyncTask;
    }

    /** * 获取 * * @param people * @return */

    public CutestPeople findCutestPeople(List<CutestPeople> people) {
        return Collections.max(people);
    }

}

在来看看我们最初逻辑部分的代码：

public class DataHelper {
    private QueryApi mApi;

    public Uri saveTheCutestPeople(String query) {
        List<CutestPeople> cutestPeoples = mApi.queryCuestPeople(query);
        CutestPeople cutestPeople = findCutestPeople(cutestPeoples);
        Uri savedUrl = mApi.store(cutestPeople);
        return savedUrl;
    }

    /** * 获取 * @param people * @return */
    public CutestPeople findCutestPeople(List<CutestPeople> people) {
        return Collections.max(people);
    }
}

对比上述的代码，是否发现逻辑是一致的，区别是我们将异步操作都细分为更小的模块，然后组合在一起，最后返回一个组合后的结果对象而已。

写了这么多，我们到底是为了阐述什么呢？这个跟RxJava有什么关系呢？不急，我们接下来使用RxJava来实现上述这个需求，如下所示：

ApiWrapper类：

public class ApiWrapper {

    private QueryApi mApi;

    public Observable<List<CutestPeople>> queryQuestPeople(final String query) {

        return Observable.create(new Observable.OnSubscribe<List<CutestPeople>>() {
            @Override
            public void call(final Subscriber<? super List<CutestPeople>> subscriber) {

                mApi.queryCutestPeople(query, new QueryApi.CutestPeopleQueryCallback() {
                    @Override
                    public void onCutestPeopleReceived(List<CutestPeople> people) {
                        subscriber.onNext(people);
                    }

                    @Override
                    public void onQueryError(Exception e) {
                        subscriber.onError(e);
                    }
                });
            }
        });
    }

    public Observable<Uri> storeUri(final CutestPeople people) {
        return Observable.create(new Observable.OnSubscribe<Uri>() {
         @Override
         public void call(final Subscriber<? super Uri> subscriber) {
                mApi.store(people, new QueryApi.StoreCallback() {
                    @Override
                    public void onCutestPeopleStore(Uri uri) {
                        subscriber.onNext(uri);
                    }

                    @Override
                    public void onStoreError(Exception e) {
                        subscriber.onError(e);
                    }
                });
            }
        });
    }
}

DataHelper类：

public class DataHelper {

    ApiWrapper mApiWrapper;

    public Observable<Uri> saveTheCuestPeople(String query) {
        Observable<List<CutestPeople>> listObservable = mApiWrapper.queryQuestPeople(query);
        Observable<CutestPeople> cutestPeople = listObservable.map(new Func1<List<CutestPeople>, CutestPeople>() {
            @Override
            public CutestPeople call(List<CutestPeople> cutestPeoples) {
                return findCutestPeople(cutestPeoples);
            }
        });
        Observable<Uri> storeObservable = cutestPeople.flatMap(new Func1<CutestPeople, Observable<Uri>>() {
            @Override
            public Observable<Uri> call(CutestPeople people) {
                return mApiWrapper.storeUri(people);
            }
        });
        return storeObservable;
    }

    /** * 获取 * * @param people * @return */

    public CutestPeople findCutestPeople(List<CutestPeople> people) {

        return Collections.max(people);
    }

}

利用lambdas表达式我们再看看这段逻辑代码：

public class DataHelper {
    ApiWrapper mApiWrapper;

    public Observable<Uri> saveTheCuestPeople(String query) {
        Observable<List<CutestPeople>> listObservable = mApiWrapper.queryQuestPeople(query);

        Observable<CutestPeople> cutestPeopleObservable = listObservable.map(peopleList -> findCutestPeople(peopleList));

        Observable<Uri> storeObservable = cutestPeopleObservable.flatMap(people -> mApiWrapper.storeUri(people));

        return storeObservable;
    }

    public CutestPeople findCutestPeople(List<CutestPeople> people) {
        return Collections.max(people);
    }
}

利用Rxjava实现的逻辑，跟我们最初实现的逻辑基本一致，而区别就是利用RxJava将一个个的异步操作单独的抽象出来，这样我们可以避免各种嵌套的回调，然后将这些抽象出来的异步操作进行组合作为一个结果返回即可。

总结：Rxjava的核心思想就是处理异步操作，将异步操作独立的抽象出来，在异步操作非常复杂的情况下，Rxjava以一条链式调用来将一系列复杂的逻辑穿成一条线，从而实现代码的简洁性与易读性。

参考文献：
[https://github.com/cn-ljb/rxjava_for_android]

更多相关文章

随机推荐