Android拾萃 - 从零打造一个RxJava（搞清楚RxJava代码为什么这么写）

任何框架都是从无到有，都是为了解决问题而产生，那么RxJava是如何产生的呢？RxJava代码的写法，为何如此让人看不懂，回调的参数等等，让小白看了摸不着头脑。

接下来的文章，主要是依据NotRxJava懒人专用指南,结合自己的理解，写的一篇更加小白的文章，以帮助我们更好的梳理和理解。

Cat 应用程序

让我们来创建一个真实世界的例子。我们都知道猫是我们技术发展的引擎，所以就让我们也来创建这么一个用来下载猫图片的典型应用吧。

任务描述

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

我们只关心下载、处理和保存猫的数据。

我们开始吧~

思维导图

这里各个点后面都会结合这个图，进行解说



模型和 API

根据我们的任务，我们能够得到的信息

1. 给定的查询信息String query

2. 搜索所有的猫图片 List queryCats(String query)

3. 每个图片包含可爱指数的参数（描述图片可爱度的整型值）找到最可爱的猫 （Cat 包含图片Bitmap image和可爱属性int cuteness）。

4. 最可爱的猫图片保存到本地（Uri store(Cat cat)，不清楚Uri 的自行查阅）

“猫”的数据结构：

public class Cat implements Comparable<Cat> {
//图片
Bitmap image;
//可爱属性
int cuteness;

//为了实现比较功能，我们的Cat对象需要实现Comparable<Cat>接口，然后重写compareTo方法
@RequiresApi(api = Build.VERSION_CODES.KITKAT)
@Override
public int compareTo(@NonNull Cat another) {
//这里大小比较只针对可爱属性  cuteness
return Integer.compare(cuteness, another.cuteness);
}
}

我们先不考虑异步情况，那么我们的接口Api：

public interface Api {
//阻塞
List<Cat> queryCats(String query);
Uri store(Cat cat);
}

然后吧我们的业务逻辑封装到我们的helper类

public class CatsHelper {
Api api;

//组合方法
public Uri saveTheCutestCat(String query){
List<Cat> cats = api.queryCats(query);
Cat cat = findCutestCat(cats);
Uri uri = api.store(cat);
return uri;
}

private Cat findCutestCat(List<Cat> cats) {
return Collections.max(cats);
}
}

我们知道，在实际应用之中，我们会处理很多任务，如果使用上面的阻塞方式，用户的每次操作都需要等待返回之后才能继续，这明显是不好的，因此我们需要异步。

异步Api接口

我们queryCats是一个http请求，搜索完毕，需要有一个回调告诉我们，以让我们处理数据，并进行下步业务逻辑，于是，新的api的代码如下：

public interface Api {
//阻塞api
List<Cat> queryCats(String query);
Uri store(Cat cat);

//异步api，这里就不需要返回值了，直接在回调里获取即可
void queryCatsAsync(String query, CatsQueryCallback callback);
void store(Cat cat, StoreCallback storeCallback);

//请求的接口回调
interface CatsQueryCallback {
void onCatListReceived(List<Cat> cats);
void onError(Exception e);
}

//保存数据的接口回调
interface StoreCallback{
void onCatStored(Uri uri);
void onStoreFailed(Exception e);
}
}

我们的helper类，作出相应的更改，为了让异步更加完整，我们比较最可爱的猫成功或者失败，都需要告诉用户，于是也需要写一个接口CutestCatCallback回调出去。

public class CatsHelper {

private Cat findCutestCat(List<Cat> cats) {
return Collections.max(cats);
}

//监听接口
public interface CutestCatCallback {
void onCutestCatSaved(Uri uri);
void onQueryFailed(Exception e);
}

Api api;

//组合方法
public Uri saveTheCutestCat(String query){
List<Cat> cats = api.queryCats(query);
Cat cat = findCutestCat(cats);
Uri uri = api.store(cat);
return uri;findCutestCat
}

//异步
public  void saveTheCutestCatAsyc(final String query, final CutestCatCallback cutestCatCallback){
api.queryCatsAsync(query, new Api.CatsQueryCallback() {
@Override
public void onCatListReceived(List<Cat> cats) {
Cat cat = findCutestCat(cats);
api.store(cat, new Api.StoreCallback() {
@Override
public void onCatStored(Uri uri) {
cutestCatCallback.onCutestCatSaved(uri);
}

@Override
public void onStoreFailed(Exception e) {
cutestCatCallback.onQueryFailed(e);
}
});

}

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

}

现在它有了更多无关代码和花括号，但是逻辑是一样的。

每一个异步操作，我们都必须创建出回调接口并在代码中手动的插入它们，接口也越来越多！

在这样的代码中错误不会自动地传递，我们需要在手动在onStoreFailed和onQueryFailed方法里面，添加代码，将错误传递给上一层。

为了解决这些问题，我们引入了泛型回调

泛型回调

泛型类

/**
* 定义泛型类
*
* 与普通类的定义相比，上面的代码在类名后面多出了 <T1, T2>，
* T1, T2 是自定义的标识符，也是参数，用来传递数据的类型，而不是数据的值，我们称之为类型参数。
* 在泛型中，不但数据的值可以通过参数传递，数据的类型也可以通过参数传递。
* T1, T2 只是数据类型的占位符，运行时会被替换为真正的数据类型。
* Created by philos on 17-9-16.
*/
public class Point<T1, T2>{
T1 x;
T2 y;
public T1 getX() {
return x;
}
public void setX(T1 x) {
this.x = x;
}
public T2 getY() {
return y;
}
public void setY(T2 y) {
this.y = y;
}
}

泛型接口

/**
* 定义泛型接口
* Created by philos on 17-9-16.
*/

interface Info<T> {
public T getVar();
}

接口实现类

/**
* 实现泛型接口
* Created by philos on 17-9-16.
*/

public class InfoImp<T> implements Info<T> {

private T var;

// 定义泛型构造方法
public InfoImp(T var) {
this.setVar(var);
}
public void setVar(T var) {
this.var = var;
}

@Override
public T getVar() {
return this.var;
}
}

泛型方法（固定参数 和 可变参数）

public class Main {
//泛型方法
public static <T> void out(T t) {
System.out.println(t);
}

//可变参数 泛型方法
public static <T> void out(T... args) {
for (T t : args) {
System.out.println(t);
}
}

public static void main(String[] args) {
//普通泛型
out("findingsea");
out(123);
out(11.11);
out(true);

//可变参数泛型
out("findingsea", 123, 11.11, true);
}
}

了解了上面具体的写法，我们继续改造我们的猫程序代码，请看上面的思维导图左边的异步优化。

泛型回调

对比接口可以发现，回调情况有两种，成功和失败，成功的返回一个对象，失败的返回Exception类，于是新增统一的回调接口如下：

/**
* 泛型回调
* 找到共同的模式，进行抽离
* 替代原来所有的接口
* Created by philos on 17-9-15.
*/

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

Api 包装类ApiWrapper

我们的之前就定义完毕了，上线之后，一般不会进行大幅度修改，这个时候，我们可以实现一个Api包装类。

/**
* 使用了泛型之后，两个接口合并为一个接口
* Created by philos on 17-9-15.
*/

public class ApiWrapper {
Api api;

public void queryCats(String query, final Callback<List<Cat>> castCallback){
api.queryCatsAsync(query, new Api.CatsQueryCallback() {
@Override
public void onCatListReceived(List<Cat> cats) {
castCallback.onResult(cats);
}

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

public void store(Cat cat, final Callback<Uri> uriCallback){
api.store(cat, new Api.StoreCallback() {
@Override
public void onCatStored(Uri uri) {
uriCallback.onResult(uri);
}

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

我们对应的helper类就不在持有Api对象了，换成了包装类，代码修改如下：

public class CatsHelper {
private Cat findCutestCat(List<Cat> cats) {
return Collections.max(cats);
}

//监听接口
public interface CutestCatCallback {
void onCutestCatSaved(Uri uri);
void onQueryFailed(Exception e);
}

//泛型回调 （retrofit很像有木有）
ApiWrapper apiWrapper;

public  void saveTheCutestCatWrap(final String query, final CutestCatCallback cutestCatCallback){
apiWrapper.queryCats(query, new Callback<List<Cat>>() {
@Override
public void onResult(List<Cat> result) {
Cat cat = findCutestCat(result);
apiWrapper.store(cat, new Callback<Uri>() {
@Override
public void onResult(Uri result) {
cutestCatCallback.onCutestCatSaved(result);
}

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

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

}

分解，使用临时泛型对象

我们发现异步操作（queryCats，queryCats，还有saveTheCutestCat），它们都遵循了相同的模式。调用它们的方法有一些参数（query、cat）也包括一个回调对象。再次说明：任何异步操作需要携带所需的常规参数和一个回调实例对象。请看思维导图左边异步优化。

临时泛型对象

对外提供的方法名暂时起为start

/**
* 分解异步操作，每步有一个参数
* 每步返回，临时对象
* 每步携带 回调信息
* 避免回调地狱，用链式解决问题
* Created by philos on 17-9-16.
*/

public abstract class AsyncJob<T> {
//包装了方法，对外隐藏，外部只需传入回调，定义返回类型，调用是start即可
public abstract void start(Callback<T> callback);
}

改造wrapper类，把我们以前的方法也隐藏起来了

/**
* 所有的异步操作都统一了
* 每次异步操作都返回AsyncJob<T>
* 每次异步都携带 回调Callback<T>
* start包装了真正的请求，外部不关心，只要调用start即可
* Created by philos on 17-9-16.
*/
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) {
//这里进行请求，然后返回给callback就可以了
api.queryCatsAsync(query, new Api.CatsQueryCallback() {
@Override
public void onCatListReceived(List<Cat> cats) {
callback.onResult(cats);
}

@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) {
//进行请求，然后返回给callback
api.store(cat, new Api.StoreCallback() {
@Override
public void onCatStored(Uri uri) {
callback.onResult(uri);
}

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

helper类也需要进行修改

public class CatsHelper {

private Cat findCutestCat(List<Cat> cats) {
return Collections.max(cats);
}

//监听接口
public interface CutestCatCallback {
void onCutestCatSaved(Uri uri);
void onQueryFailed(Exception e);
}
//分解统一返回临时对象,链式调用，高大上有木有

ApiWrapper apiWrapper;

public AsyncJob<Uri> saveTheCutestCat(final String query){
//直接new 需要返回的对象
return new AsyncJob<Uri>() {
@Override
public void start(final Callback<Uri> callback) {
//这里请求数据，返回,  每个异步都有自己的回调信息这里重新new 一个CallBack
apiWrapper.queryCats(query).start(new Callback<List<Cat>>() {
@Override
public void onResult(List<Cat> result) {
//进行数据操作和转换
Cat cat = findCutestCat(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);
}
});
}
};
}
}

是我们的逻辑数据流：

(async)                 (sync)           (async)
query ===========> List<Cat> -------------> Cat ==========> Uri
queryCats              findCutest           store

分解成更小的操作

public class CatsHelper {

private Cat findCutestCat(List<Cat> cats) {
return Collections.max(cats);
}

//监听接口
public interface CutestCatCallback {
void onCutestCatSaved(Uri uri);
void onQueryFailed(Exception e);
}

ApiWrapper apiWrapper;

public AsyncJob<Uri> saveTheCutestCat(String query) {
//获取猫列表  返回临时对象  catsListAsyncJob
final AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper.queryCats(query);

//查找最可爱猫  返回临时对象  cutestCatAsyncJob
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(findCutestCat(result));
}

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

//保存猫到本地  返回临时对象  storedUriAsyncJob
AsyncJob<Uri> storedUriAsyncJob = new AsyncJob<Uri>() {
@Override
public void start(final Callback<Uri> cutestCatCallback) {
cutestCatAsyncJob.start(new Callback<Cat>() {
@Override
public void onResult(Cat cutest) {
apiWrapper.store(cutest)
.start(new Callback<Uri>() {
@Override
public void onResult(Uri result) {
cutestCatCallback.onResult(result);
}

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

@Override
public void onError(Exception e) {
cutestCatCallback.onError(e);
}
});
}
};
return storedUriAsyncJob;
}

映射 (其实就是操作符的封装)

现在看看 AsyncJob cutestCatAsyncJob的部分：

AsyncJob<Cat> cutestCatAsyncJob = new AsyncJob<Cat>() {
@Override
public void start(Callback<Cat> callback) {
catsListAsyncJob.start(new Callback<List<Cat>>() {
@Override
public void onResult(List<Cat> result) {
callback.onResult(findCutest(result));
}

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

这 16 行代码只有一行是对我们有用（对于逻辑来说）的操作：

findCutest(result)

剩下的仅仅是开启另外一个AsyncJob和传递结果与错误的样板代码。此外，这些代码并不用于特定的任务，我们可以把其移动到其它地方而不影响编写我们真正需要的业务代码。

我们要怎么做呢？

将方法findCutest分离出来，并和前面一样返回临时对象，供下步使用。

在 Java 中不能直接传递方法（函数）所以我们需要通过类（和接口）来间接实现这样的功能，我们来定义转换方法的接口：

转换方法的接口

/**
* 方法传入接口
* T对应于参数类型而R对应于返回类型
* Created by philos on 17-9-16.
*/

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

我们之前返回的临时对象的start方法是通过回调返回的，但是现在是调用方法转换需要返回临时对象，接着上面继续看怎么改造。

我们新增一个方法，就叫map，返回对象R，参考上面泛型方法，AsyncJob代码修改如下：

public abstract class AsyncJob<T> {
//包装了方法，对外隐藏，外部只需传入回调，定义返回类型，调用是start即可
public abstract void start(Callback<T> callback);
//返回临时对象 方法(需要传入映射的方法Func<T, R> func携带的是对象R)，
public <R> AsyncJob<R> map(final Func<T, R> func){
//使用自己，为链式调用返回同样的对象（return 一个新建的）
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) {
//方法 替换成func
//返回对象为R
R mapped = func.call(result);
callback.onResult(mapped);
}

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

前面map对应的是分离一个方法返回的是实体对象的情况，那么如果是需要返回同样的临时对象的方法，比如上面的CatsHelper 的storedUriAsyncJob嵌套了两个start方法，所以AsyncJob在添加一个方法，就叫做flatMap

//返回临时对象 方法(需要传入映射的方法Func<T, AsyncJob<R>> func携带的是上一个临时对象AsyncJob<R>)
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(final T result) {
//传入的方法不一样，返回值不一样，在这里有区别
AsyncJob<R> mapped = func.call(result);
mapped.start(new Callback<R>() {
@Override
public void onResult(R result) {
callback.onResult(result);
}

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

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

这里我们可以简单的理解，我们之前把回调用start表示，但是对于start还调用了其他方法的，还是存在嵌套，那么我们的map就是为了解决这个问题的。但是如果是多次start的嵌套，map明显不满足，这个时候我们使用flatMap 。

于是，我们的helper代码修改如下：

public class CatsHelper {

private Cat findCutestCat(List<Cat> cats) {
return Collections.max(cats);
}

//监听接口
public interface CutestCatCallback {
void onCutestCatSaved(Uri uri);
void onQueryFailed(Exception e);
}
//映射
public AsyncJob<Uri> saveTheCutestCat4(String query){
//获取猫列表  返回临时对象  catsListAsyncJob
final AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper2.queryCats(query);

//查找最可爱猫  返回临时对象  cutestCatAsyncJob
//这里用Func将方法findCutestCat(result)分离出来，并返回一个临时对象cutestCatAsyncJob
final AsyncJob<Cat> cutestCatAsyncJob = catsListAsyncJob.map(new Func<List<Cat>, Cat>() {
@Override
public Cat call(List<Cat> cats) {
return findCutestCat(cats);
}
});

//保存猫到本地  返回临时对象  storedUriAsyncJob
AsyncJob<AsyncJob<Uri>> storedUriAsyncJob = cutestCatAsyncJob.map(new Func<Cat, AsyncJob<Uri>>() {
@Override
public AsyncJob<Uri> call(Cat cat) {
return apiWrapper2.store(cat);
}
});
return storedUriAsyncJob;
}
}

RxJava

嘿，你不需要把那些代码拷到你的项目中，因为我们还是实现地不够完全的，仅仅算是非线程安全的 RxJava 的一小部分而已。

它们之间只有一些差异：

AsyncJob

就是实际上的 Observable

，它不仅可以只分发一个单一的结果也可以是一个序列（可以为空）。

Callback

就是 Observer

，除了 Callback 少了onNext(T t)

方法。Observer 中在onError(Throwable t)

方法被调用后，会继而调用onCompleted()

，然后 Observer 会包装好并发送出事件流（因为它能发送一个序列）。

abstract void start(Callback callback)

对应 Subscription subscribe(final Observer