RX前传

原型

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

模型和API

猫的结构数据

1
2
3
4
5
6
7
8
9
public class Cat implements Comparable<Cat>{
	Bitmap images;
	int cuteness;
	
	@Override
	public int compareTo(Cat another){
		return Integer.compare(cuteness, another.cuteness);
	}
}

阻塞式API

1
2
3
4
public interface Api{
	List<Cat> queryCats(String query);
	Uri store(Cat cat);
}

业务逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class CatHelper{
	Api api;
	
	public Uri saveTheCutestCat(String query){
		List<cat> cats = api.queryCats(query);
		Cat cutest = findCutest(cats);
		Uri savedUri = api.store(cutest);
		return savedUri;
	}
	
	private Cat findCutest(List<Cat> cats){
		return Collections.max(cats);
	}
}

主方法savetheCutestCat只包含了3个其他方法， 提供了输入参数然后就能得到结果返回了，
在这个方法工作的时候我们需要等待它的完成。

优势：

组合：根据其他３个方法而新创建一个方法(saveTheCutestCat),因此组合了它们。
错误的传递：另外一个好处就是我们处理错误的方式，任何一个方法都可能因执行时发生错误而被终止，这个错误能在任何层次上被处理掉，Java中我们叫它抛异常。我们不需要为组合方法里的每个方法都做异常处理，仅需要对这些组合起来的方法做统一处理
1
2
3
4
5
6
7
8
try{
	List<Cat> cats = api.queryCats(query);
	Cat cutest = findCutest(cats);
	Uri savedUri = api.store(cutest);
}catch(Exception e){
	e.printStackTrace();
	return someDefaultValue;
}




走向异步

异步的网络调用

1
2
3
4
5
6
7
8
9
public interface Api{
	interface CatsQueryCallback{
		void onCatListReceived(List<Cat> cats);
		void onError(Exception e);
	}
	
	void queryCats(String query,CatsQueryCallback catsQueryCallback);
	Uri store(Cat cat);
}

异步的获取猫的信息集合列表，返回正确或错误的结果时都会通过CatsQueryCallback回调接口。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public calss CatsHelper{
	public interface CutestCatCallback{
		void onCutestCatSaved(Uri uri);
		void onQueryFailed(Exception e);
	}
	
	Api api;
	
	public void saveTheCutestCat(String query, CutestCatCallback cutestCatCallback){
		api.queryCats(query, new Api.CatsQueryCallback(){
			@Override
			public void onCatListReceived(List<Cat> cats){
				Cat cutest = findCutest(cats);
				Uri savedUri = api.store(cutest);
				cutestCatCallback.onCutestCatSaved(savedUri);
			}
			@Override
			public void onQueryFailed(Exception e){
				cutestCatCallback.onError(e);
			}
		});
	}
	
	private Cat findCutest(List<Cat> cats){
		return Collections.max(cats);
	}
}

两个Api方法都调用异步

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public interface Api{
	interface CatsQueryCallback{
		void onCatListReceived(List<Cat> cats);
		void onQueryFailed(Exception e);
	}
	
	interface StoreCallback{
		void onCatStored(Uri uri);
		void onStoreFailed(Exception e);
	}
	
	void queryCats(String query, CatsQueryCallback catsQueryCallback);
	void store(Cat cat, StoreCallback storeCallback);
}

helper

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
public class CatsHelper{
	public interface CutestCatCallback{
		void onCutestCatSaved(Uri uri);
		void onError(Exception e);
	}
	
	Api api;
	
	public void saveTheCutestCat(String query, CutestCatCallback cutestCatCallback){
		api.queryCats(query, new Api.CatsQueryCallback(){
			@Override
			public void onCatListReceived(List<Cat> cats){
				Cat cutest = findCutest(cats);
				api.store(cutest, new Api.StoreCallback(){
					@Override
					public void onCatStored(Uri uri){
						cutestCatCallback.onCutestCatSaved(uri);
					}
					
					@Override
					public void onStoreFailed(Exception e){
						cutestCatCallback.onCutestCatSaved(uri);
					}
				});
			}
			
			@Override
			public void onQueryFailed(Exception e){
				cutestCatCallback.onError(e);
			}
		});
	}
	
	private Cat findCutest(List<Cat> cats){
		return Collections.max(cats);
	}
	
}




泛型回调

从回调接口中找到共同的模式：

  • 都有一个分发结果的方法(onCutestCatSaved, onCatListReceived, onCatStored)
  • 它们中大多数有一个用于错误处理的方法(onError, onQueryFailed, onStoreFailed)
    所以可以创建一个泛型回调接口去代替原来所有的接口。
    但是不能去改变API的调用方法的签名，必须用创建包装类来间接调用。
    1
    2
    3
    4
    public interface Callback<T>{
    	void onResult(T result);
    	void onError(Exception e);
    }

ApiWrapper

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
public class ApiWrapper{
	Api api;
	
	public void queryCats(String query, Callback<List<Cat>> catsCallback){
		api.queryCats(query, new Api.CatsQueryCallback(){
			@Override
			public void onCatListReceived(List<Cat> cats){
				catsCallback.onResult(cats);
			}
			
			@Override
			public void onQueryFailed(Exception e){
				catsCallback.onError(e);
			}
		}
	}
	
	public void store(Cat cat, 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);
			}
		});
	}
}

CatsHelper

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public class CatsHelper{
	ApiWrapper apiWrapper;
	
	public void saveTheCutestCat(String query, Callback<Uri> cutestCatCallback){
		apiWrapper.queryCats(query, new Callback<List<Cat>> ()){
			@Override
			public void onResult(List<Cat> cats){
				Cat cutest = findCutest(cats);
				apiWrapper.store(cutest, cutestCatCallback);
			}
			
			@Override
			public void onError(Exception e){
				cutestCatCallback.onError(e);
			}
		});
	}
}




做的更好

如果在异步操作中返回一些临时对象，需要定义一个出来。
这样的一个对象需要包括常见的行为（以回调为单一参数），定义这样的类给所有的异步操作使用，这个类叫它AsyncJob。

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

更改包装API的调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
public class ApiWrapper{
	Api api;
	
	public AsyncJob<List<Cat>> queryCats(String query){
		return new AsyncJob<List<Cat>>(){
			@Override
			public void start(Callback<List<Cat>> catsCallback){
				api.queryCats(query, new Api.CatsQueryCallback(){
					@Override
					public void onCatListReceived(List<Cat> cats){
						catsCallback.onResult(cats);
					}
					
					@Override
					public void onQueryFailed(Exception e){
						catsCallback.onError(e);
					}
				});
			}
		};
	}
	
	public AsyncJob<Uri> store(Cat cat){
		return new AsyncJob<Uri>(){
			@Override
			public void start(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);
					}
				});
			}
		};
	}
	
}

CatsHelper

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
public class CatsHelper{
	ApiWrapper apiWrapper;
	
	public AsyncJob<Uri> saveTheCutestCat(String query){
		return new AsyncJob<Uri>(){
			@Override
			public void start(Callback<Uri> cutestCatCallback){
				apiWrapper.queryCats(query)
					.start(new Callback<List<Cat>>()){
						@Override
						public void onResult(List<Cat> cats){
							Cat cutest = findCutest(cats);
							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);
						}
					});
			}
		};
	}
	
	private Cat findCutest(List<Cat> cats){
		return Collections.max(cats);
	}
}

现在可以给客户端返回“组合”操作的AsyncJob。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
public class CatsHelper{
	ApiWrapper apiWrapper;
	
	public AsyncJob<Uri> saveTheCutestCat(String query){
		AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper.queryCats(query);
		AsyncJob<Cat> catsListAsyncJob = 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);
					}
				});
				
				
			}
			
		};
		
		AsyncJob<Uri> storedUriAsyncJob = new AsyncJob<Uri>(){
			@Override
			public void start(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;
	}
	
	private Cat findCutest(List<Cat> cats){
		return Collections.max(cats);
	}

}

AsyncJob cutestCatAsyncJob

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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行代码只有一行有用（对于逻辑来说）

1
findCutest(result);

剩下的仅仅是开启另外一个AsyncJob和传递结果与错误的样板代码。

该怎么写？必须做下面两件事情:

  • AsyncJob是我们转换的结果
  • 转换方法
    1
    2
    3
    public interface Func<T, R>{
    	R call(T t);
    }

相当简单，Func接口有两个类型成员，T对应于参数类型而R对应于返回类型。
当从一个AsyncJob中装换处结果后就需要做一些值之间的映射，这样的方法叫map。
定义这个方法实例（Func类型）最好的地方就在AsyncJob类中，所以AsyncJob代码里看起来就是这样了：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public abstract class AsyncJob<T>{
	public abstract void start(Callback<T> callback);
	
	public <R> AsyncJob<R> map(Func<T, R> func){
		final AsyncJob<T> source = this;
		return new AsyncJob<R>(){
			@Override
			public void start(Callback<R> callback){
				source.start(new Callback<T>(){
					@Override
					public void onResult(T result){
						R mapped = func.call(result);
						callback.onResult(mapped);
					}
					
					@Override
					public void onError(Exception e){
						callback.onError(e);
					}
				});
			}
		};
	}
}

这时CatsHelper就是下面这样了：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
public class CatsHelper{
	ApiWrapper apiWrapper;
	
	public AsyncJob<Uri> saveTheCutestCat(String query){
		AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper.queryCats(query);
		AsyncJob<Cat> cutestCatAsyncJob = catsListAsyncJob.map(new Func<List<Cat>, Cat>(){
			@Override
			public Cat call(List<Cat> cats){
				return findCutest(cats);
			}
		});
		
		AsyncJob<Uri> storedUriAsyncJob = cutestCatAsyncJob.flatMap(new Func<Cat, AsyncJob<Uri>(){
			@Override
			public void start(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;
		}
	}
	
	private Cat findCutest(List<Cat> cats){
		return Collections.max(cats);
	}
}




高级映射

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public class CatsHelper {
 
    ApiWrapper apiWrapper;
 
    public AsyncJob<Uri> saveTheCutestCat(String query) {
        AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper.queryCats(query);
        AsyncJob<Cat> cutestCatAsyncJob = catsListAsyncJob.map(new Func<List<Cat>, Cat>() {
            @Override
            public Cat call(List<Cat> cats) {
                return findCutest(cats);
            }
        });
 
        AsyncJob<Uri> storedUriAsyncJob = cutestCatAsyncJob.map(new Func<Cat, Uri>() {
            @Override
            public Uri call(Cat cat) {
                return apiWrapper.store(cat);
        //      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ will not compile
        //      Incompatible types:
        //      Required: Uri
        //      Found: AsyncJob<Uri>                
            }
        });
        return storedUriAsyncJob;
    }
 
    private Cat findCutest(List<Cat> cats) {
        return Collections.max(cats);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public class CatsHelper {

    ApiWrapper apiWrapper;

    public AsyncJob<Uri> saveTheCutestCat(String query) {
        AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper.queryCats(query);
        AsyncJob<Cat> cutestCatAsyncJob = catsListAsyncJob.map(new Func<List<Cat>, Cat>() {
            @Override
            public Cat call(List<Cat> cats) {
                return findCutest(cats);
            }
        });

        AsyncJob<AsyncJob<Uri>> storedUriAsyncJob = cutestCatAsyncJob.map(new Func<Cat, AsyncJob<Uri>>() {
            @Override
            public AsyncJob<Uri> call(Cat cat) {
                return apiWrapper.store(cat);
            }
        });
        return storedUriAsyncJob;
        //^^^^^^^^^^^^^^^^^^^^^^^ will not compile
        //      Incompatible types:
        //      Required: AsyncJob<Uri>
        //      Found: AsyncJob<AsyncJob<Uri>>
    }

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

在目前这点上我们只能有AsyncJob>。我们需要往更深处挖吗？我们希望的是，去把AsyncJob在一个级别上的两个异步操作扁平化成一个单一的异步操作。

现在我们需要的是得到能使方法返回映射成R类型也是AsyncJob类型的操作。这个操作应该像map，但在最后应该flatten我们嵌套的AsyncJob。我们叫它为flatMap

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
public abstract class AsyncJob<T>{
	public abstract void start(Callback<T> callback);
	
	public <R> AsyncJob<R> map(Func<T,R> func){
		final AysncJob<T> source = this;
		return new AsyncJob<R>(){
			@Override
			public void start(Callback<R> callback){
				@Override
				public void start(Callback<R> callback){
					source.start(new Callback<T>(){
						@Override
						public void onResult(T result){
							R mapped = func.call(result);
							callback.onResult(mapped);
						}
						
						@Override
						public void onError(Exception e){
							callback.onError(e);
						}
					});
				}
			}
		};
	}
	
	public <R> AsyncJob<R> flatMap(Func<T,AsyncJob<R>> func){
		final AsyncJob<T> source = this;
		return new AsyncJob<R>(){
			@Override
			public void start(Callback<R> callback){
				source.start(new Callback<T>(){
					@Override
					public void onResult(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);
					}
				});
			}
		};
	}
}

修复CatsHelper

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public class CatsHelper{
	ApiWrapper apiWrapper;
	
	public AsyncJob<Uri> saveTheCutestCat(String query){
		AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper.queryCats(query);
		AsyncJob<Cat> cutestCatAsyncJob = catsListAsyncJob.map(new Func<List<Cat>, Cat>(){
			@Override
			public Cat call(List<Cat> cats){
				return findCutest(cats);
			}
		});
		
		AsyncJob<Uri> storeUriAsyncJob = cutestCatAsyncJob.flatMap(new Func<Cat, AsyncJob<Uri>>(){
			@Override
			public AsyncJob<Uri> call(Cat cat){
				return apiWrapper.store(cat);
			}
		});
		
		return storedUriAsyncJob;
	}
	
	private Cat findCutest(List<Cat> cats){
		return Collections.max(cats);
	}
}




最后的要点

如果使用Java 8的lambda表达式，代码会更加简洁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class CatsHelper{
	ApiWrapper apiWrapper;
	
	public AsyncJob<Uri> saveTheCutestCat(String query){
		AsyncJob<List<Cat>> catsListAsyncJob = apiWrapper.queryCats(query);
		AsyncJob<Cat> cutestCatAsyncJob = catsListAsyncJob.map(cats->findCutest(cats));
		AsyncJob<Uri> storedUriAsyncJob = cutestCatAsyncJob.flatMap(cat->apiWrapper.store(cat));
		return storedUriAsyncJob;
	}
	
	private Cat findCutest(List<Cat> cats){
		return Collections.max(cats);
	}
	
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public class CatsHelper {
 
    Api api;
 
    public Uri saveTheCutestCat(String query){
        List<Cat> cats = api.queryCats(query);
        Cat cutest = findCutest(cats);
        Uri savedUri = api.store(cutest);
        return savedUri;
    }
 
    private Cat findCutest(List<Cat> cats) {
        return Collections.max(cats);
    }
}