RxJava前世今生

需求

聚合数据API提供了搜索某个城市未来一周每天的天气详细信息，包含最低最高气温、风速、湿度等数据。我们的任务就是：获取南京市未来一周的天气情况，找出温度最高的那一天的天气详细信息，把这条信息保存在数据库中。

同步方法实现

我们只关注获取数据、处理数据和保存数据的三个核心功能。
Model 和 API
简单的描述一下Weather的数据模型

public class Weather implements Comparable<Weather> {

    private int max;//最高温度
    private int min;//最低温度
    private String wind;

    @Override
    public int compareTo(@NonNull Weather weather) {
        return Integer.compare(max,weather.max);
    }
}

下面是我们请求数据和存储数据的API（同步），我们暂且忽略具体的实现

public interface Api {
    List<Weather> queryWeathers(String city);
    Uri save(Weather weather);
}

现在，开始编写我们的业务逻辑代码

public class WeathersHelper {
    Api api;
    public Uri saveTheMaxDay(String city){
        List<Weather> weathers = api.queryWeathers(city);
        Weather max = findMaxDay(weathers);
        Uri savedUri = api.save(max);
        return savedUri;
    }
 
    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

(@ο@) 哇～，这也太简单太清晰了吧！来看看上面的代码是多么的酷。主要的函数 saveTheMaxDay 只包含了 3个函数调用。使用参数来调用这些函数，并接收返回的参数。并且等待每个函数执行并返回结果。如此简单、如此有效。下面来看看这种简单函数的其他优点。

组合（Composition）
可以看到我们的 saveTheMaxDay 由其他三个函数调用所组成的。我们通过函数来把一个大功能分割为每个容易理解的小功能。通过函数调用来组合使用这些小功能。使用和理解起来都相当简单。
异常传递
另外一个使用函数的好处就是方便处理异常。每个函数都可以通过抛出异常来结束运行。该异常可以在抛出异常的函数里面处理，也可以在调用该函数的外面处理，所以我们无需每次都处理每个异常，我们可以在一个地方处理所有可能抛出的异常。

try{
    List<Weather> weathers = api.queryWeathers(city);
    Weather max = findMaxDay(weathers);
    Uri savedUri = api.save(max);
    return savedUri;
} catch (Exception e) {
    e.printStackTrace()
    return someDefaultValue;
}

这样，我们就可以处理这三个函数中所抛出的任何异常了。如果没有 try catch 语句，我们也可以把异常继续传递下去。

异步回调方法实现

但是，现实世界中我们往往没法等待。有些时候你没法只使用阻塞调用。在 Android 中你需要处理各种异步操作。
就拿 Android 的 OnClickListener 接口来说吧，如果你需要处理一个 View 的点击事件，你必须提供一个该 Listener 的实现来处理用户的点击事件。下面来看看如何处理异步调用。
现在我们把请求网络放在异步操作中，使用回调获取结果

public interface Api {
    interface WeathersQueryCallback {
        void onWeatherListReceived(List<Weather> weathers);
        void onError(Exception e);
    }
    void queryWeathers(String city, WeathersQueryCallback weathersQueryCallback);
    Uri save(Weather weather);
}

这样我们查询天气的操作就变为异步的了，通过 WeathersQueryCallback 回调接口来结束查询的数据和处理异常情况。
我们的业务逻辑也需要跟着改变一下：

public class WeathersHelper {
 
    public interface MaxDayCallback {
        void onMaxDaySaved(Uri uri);
        void onQueryFailed(Exception e);
    }
    Api api;
    public void saveTheMaxDay(String city, MaxDayCallback maxDayCallback){
        api.queryWeathers(city, new Api.WeathersQueryCallback() {
            @Override
            public void onWeatherListReceived(List<Weather> weathers) {
                Weather max = findMaxDay(weathers);
                Uri savedUri = api.save(max);
                maxDayCallback.onMaxDaySaved(savedUri);
            }
 
            @Override
            public void onError(Exception e) {
                maxDayCallback.onQueryFailed(e);
            }
        });
    }
 
    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

这样我们就没法使用阻塞式函数调用了，我们无法继续使用阻塞调用来使用这些 API 了。所以，我们没法让 saveTheMaxDay 函数返回一个值了，我们需要一个回调接口来异步的处理结果。
这里我们再进一步，使用两个异步操作来实现我们的功能，例如我们还使用异步 IO 来写文件。

public interface Api {
    interface WeathersQueryCallback {
        void onWeatherListReceived(List<Weather> weathers);
        void onError(Exception e);
    }
    interface SaveCallback{
        void onWeatherSaved(Uri uri);
        void onSaveFailed(Exception e);
    }
    void queryWeathers(String city, WeathersQueryCallback weathersQueryCallback);
    void save(Weather weather, SaveCallback saveCallback);
}

业务逻辑变为：

public class WeathersHelper {
 
    public interface MaxDayCallback {
        void onMaxDaySaved(Uri uri);
        void onQueryFailed(Exception e);
    }
 
    Api api;
 
    public void saveTheMaxDay(String city, MaxDayCallback maxDayCallback){
        api. queryWeathers(city, new Api.WeathersQueryCallback() {
            @Override
            public void onWeatherListReceived(List< Weather > weathers) {
                Weather max = findMaxDay(weathers);
                api.save(max, new Api.SaveCallback() {
                    @Override
                    public void onWeatherSaved(Uri uri) {
                        maxDayCallback.onMaxDaySaved(uri);
                    }
 
                    @Override
                    public void onStoreFailed(Exception e) {
                        maxDayCallback.onError(e);
                    }
                });
            }
 
            @Override
            public void onQueryFailed(Exception e) {
                maxDayCallback.onError(e);
            }
        });
    }
 
    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

现在再来看看我们的业务逻辑代码，是不是和之前的阻塞式调用那么简单、那么清晰？当然不一样了，上面的异步操作代码看起来太恐怖了！这里有太多的干扰代码了，太多的匿名类了，但是不可否认，他们的业务逻辑其实是一样的。查询天气的列表数据、找出最温度最高的那一天。
组合功能也不见了！现在你没法像阻塞操作一样来组合调用每个功能了。异步操作中，每次你都必须通过回调接口来手工的处理结果。
那么关于异常传递和处理呢？没了！异步代码中的异常不会自动传递了，我们需要手工的重新传递出去。（onSaveFailed 和 onQueryFailed 就是干这事用的）
上面的代码非常难懂也更难发现潜在的 BUG。

代码优化

泛型接口
仔细的看看这些回调接口，你会发现一个通用的模式：
1，这些接口都有一个函数来返回结果（onMaxDaySaved, onWeatherListReceived, onWeatherSaved）。
2，这里还都有一个返回异常情况的函数（onError, onQueryFailed, onSaveFailed）。

所以我们可以使用一个泛型接口来替代这三个接口。由于我们无法修改 API 调用的参数类型，必须要创建一个包装类来调用泛型接口。新的接口定义如下：

public interface Callback<T> {
    void onResult(T result);

    void onError(Exception e);
}

然后使用 ApiWrapper 来改变调用的参数。

public class ApiWrapper {
    Api api;

    public void queryWeathers(String city, final Callback<List<Weather>> weathersCallback) {
        api.queryWeathers(city, new Api.WeathersQueryCallback() {
            @Override
            public void onWeatherListReceived(List<Weather> weathers) {
                weathersCallback.onResult(weathers);
            }

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

    public void save(Weather weather, final Callback<Uri> uriCallback) {
        api.save(weather, new Api.SaveCallback() {
            @Override
            public void onWeatherSaved(Uri uri) {
                uriCallback.onResult(uri);
            }

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

上面的代码使用同样的逻辑在 Callback 接口中来处理结果和异常情况。
最后 WeathersHelper 的代码如下：

public class WeatherHelper {

    ApiWrapper apiWrapper;

    public void saveTheMaxDay(String city, final Callback<Uri> maxDayCallback){
        apiWrapper.queryWeathers(city, new Callback<List<Weather>>() {

            @Override
            public void onResult(List<Weather> result) {
                Weather maxDay = findMaxDay(result);
                apiWrapper.save(maxDay,maxDayCallback);
            }

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

    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

由于使用了泛型回调接口，这里的maxDayCallback可以直接设置为函数 apiWrapper.save的参数，所以上面的代码比前面的代码要少一层匿名类。看起来简单一点。

分离参数和回调接口
看看这些新的异步操作（queryWeathers, save和 saveTheMaxDay）。这些函数都有同样的模式。使用一些参数来调用这些函数（city,weather），同时还有一个回调接口作为参数。甚至，所有的异步操作都带有一些常规参数和一个额外的回调接口参数。如果我们把他们分离开会如何，让每个异步操作只有一些常规参数而把返回一个临时的对象来操作回调接口。下面来试试看看这种方式能否有效。
如果我们返回一个临时的对象作为异步操作的回调接口处理方式，我们需要先定义这个对象。由于对象遵守通用的行为（有一个回调接口参数），我们定义一个能用于所有操作的对象。我们称之为 AsyncJob。

注意：我非常想把这个名字称之为 AsyncTask。但是由于 Android 系统已经有个 AsyncTask 类了，为了避免混淆，所以就用 AsyncJob了。
该对象如下：

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

start 函数有个Callback 回调接口参数，并开始执行该操作。
ApiWrapper 修改为：

public class ApiWrapper {
    Api api;

    public AsyncJob<List<Weather>> queryWeathers(final String city) {
        return new AsyncJob<List<Weather>>() {
            @Override
            public void start(final Callback<List<Weather>> weathersCallback) {
                api.queryWeathers(city, new Api.WeathersQueryCallback() {
                    @Override
                    public void onWeatherListReceived(List<Weather> weathers) {
                        weathersCallback.onResult(weathers);
                    }

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

    public AsyncJob<Uri> save(final Weather weather) {
        return new AsyncJob<Uri>() {
            @Override
            public void start(final Callback<Uri> uriCallback) {
                api.save(weather, new Api.SaveCallback() {
                    @Override
                    public void onWeatherSaved(Uri uri) {
                        uriCallback.onResult(uri);
                    }

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

目前看起来还不错。现在可以使用 AsyncJob 来启动每个操作了。这些功能在 WeathersHelper 中：

public class WeatherHelper {
    ApiWrapper apiWrapper;

    public AsyncJob<Uri> saveTheMaxDay(final String city) {
        return new AsyncJob<Uri>() {
            @Override
            public void start(final Callback<Uri> maxDayCallback) {
                apiWrapper.queryWeathers(city)
                        .start(new Callback<List<Weather>>() {
                            @Override
                            public void onResult(List<Weather> result) {
                                Weather maxDay = findMaxDay(result);
                                apiWrapper.save(maxDay)
                                        .start(new Callback<Uri>() {
                                            @Override
                                            public void onResult(Uri result) {
                                                maxDayCallback.onResult(result);
                                            }

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

                            @Override
                            public void onError(Exception e) {

                            }
                        });
            }
        };
    }
    
    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

看起来比前面一个版本更加复杂啊，这样有啥好处啊？
这里其实我们返回的是一个 AsyncJob 对象，该对象和客户端代码组合使用，这样在 Activity或者 Fragment 客户端代码中就可以操作这个返回的对象了。
代码虽然目前看起来比较复杂，下面我们就来改进一下。

分解
下面是数据流图：

         (async)                   (sync)               (async)
query ===========> List<Weather> -------------> Weather==========> Uri
       queryWeathers                findMax               save

为了让代码具有可读性，我们把这个流程分解为每个操作。同时我们再进一步假设，如果一个操作是异步的，则每个调用该异步操作的函数也是异步的。例如：如果查询天气是个异步操作，则找到最最高温度的那一天也是异步的。
因此，我们可以使用 AsyncJob 来把每个操作分解为一个非常小的函数里面去。

public class WeatherHelper {
    ApiWrapper apiWrapper;

    public AsyncJob<Uri> saveTheMaxDay(final String city) {
        //第一步,请求数据
        final AsyncJob<List<Weather>> weatherListAsyncJob = apiWrapper.queryWeathers(city);
        //第二步,处理数据
        final AsyncJob<Weather> maxDayAsyncJob = new AsyncJob<Weather>() {
            @Override
            public void start(final Callback<Weather> callback) {
                weatherListAsyncJob.start(new Callback<List<Weather>>() {
                    @Override
                    public void onResult(final List<Weather> result) {
                        Weather maxDay = findMaxDay(result);
                        callback.onResult(maxDay);
                    }

                    @Override
                    public void onError(Exception e) {
                        callback.onError(e);
                    }
                });
            }
        };
        //第三步,存储数据
        AsyncJob<Uri> savedUriAsyncJob = new AsyncJob<Uri>() {
            @Override
            public void start(final Callback<Uri> callback) {
                maxDayAsyncJob.start(new Callback<Weather>() {
                    @Override
                    public void onResult(Weather result) {
                        apiWrapper.save(result)
                                .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) {

                    }
                });
            }
        };

        return savedUriAsyncJob;
    }

    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

虽然代码量多了，但是看起来更加清晰了。嵌套的回调函数没那么多层级了，异步操作的名字也更容易理解了（weatherListAsyncJob , maxDayAsyncJob , savedUriAsyncJob ）。
看起来还不错，但是还可以更好。

简单的映射
先来看看我们创建 AsyncJob maxDayAsyncJob 的代码：

AsyncJob<Weather> maxDayAsyncJob = new AsyncJob<Weather>() {
            @Override
            public void start(final Callback<Weather> callback) {
                weatherListAsyncJob.start(new Callback<List<Weather>>() {
                    @Override
                    public void onResult(final List<Weather> result) {
                        Weather maxDay = findMaxDay(result);
                        callback.onResult(maxDay);
                    }

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

这 16 行代码中，只有一行代码是我们的业务逻辑代码：findMaxDay(result)
其他的代码只是为了启动 AsyncJob 并接收结果和处理异常的干扰代码。但是这些代码是通用的，我们可以把他们放到其他地方来让我们更加专注业务逻辑代码。
那么如何实现呢？需要做两件事情：
1，转换 AsyncJob 的结果
2，转换的函数
但是由于 Java 的限制，无法把函数作为参数，所以需要用一个接口（或者类）并在里面定义一个转换函数：

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

该接口很简单。有两个泛型类型定义， T 代表参数的类型； R 代表返回值的类型。
当我们把 AsyncJob 的结果转换为其他类型的时候，我们需要把一个结果值映射为另外一种类型，这个操作我们称之为 map。把该函数定义到 AsyncJob 类中比较方便，这样就可以通过 this 来访问 AsyncJob 对象了。

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

    /**
     * 简单映射
     * @param func
     * @param <R> 输入类型
     * @return 输出类型
     */
    public <R> AsyncJob<R> map(final Function<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) {
                        R mapped = func.call(result);
                        callback.onResult(mapped);
                    }

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

看起来不错，现在的 CatsHelper 如下：

public class WeatherHelper {
    ApiWrapper apiWrapper;

    public AsyncJob<Uri> saveTheMaxDay(final String city) {
        //第一步,请求数据
        final AsyncJob<List<Weather>> weatherListAsyncJob = apiWrapper.queryWeathers(city);
        //第二步,处理数据
        final AsyncJob<Weather> maxDayAsyncJob = weatherListAsyncJob.map(new Function<List<Weather>, Weather>() {
            @Override
            public Weather call(List<Weather> weathers) {
                return findMaxDay(weathers);
            }
        });
        //第三步,存储数据
        AsyncJob<Uri> savedUriAsyncJob = new AsyncJob<Uri>() {
            @Override
            public void start(final Callback<Uri> callback) {
                maxDayAsyncJob.start(new Callback<Weather>() {
                    @Override
                    public void onResult(Weather result) {
                        apiWrapper.save(result)
                                .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) {

                    }
                });
            }
        };

        return savedUriAsyncJob;
    }

    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

新的创建 AsyncJob maxDayAsyncJob 的代码只有 6行，并且只有一层嵌套。

高级映射
但是 AsyncJob storedUriAsyncJob 看起来还是非常糟糕。这里也能使用映射吗？下面就来试试吧！

public class WeatherHelper {
    ApiWrapper apiWrapper;

    public AsyncJob<Uri> saveTheMaxDay(final String city) {
        //第一步,请求数据
        final AsyncJob<List<Weather>> weatherListAsyncJob = apiWrapper.queryWeathers(city);
        //第二步,处理数据
        final AsyncJob<Weather> maxDayAsyncJob = weatherListAsyncJob.map(new Function<List<Weather>, Weather>() {
            @Override
            public Weather call(List<Weather> weathers) {
                return findMaxDay(weathers);
            }
        });
        //第三步,存储数据
        AsyncJob<Uri> savedUriAsyncJob = maxDayAsyncJob.map(new Function<Weather, Uri>() {
            @Override
            public Uri call(Weather weather) {
                return apiWrapper.save(weather);
        //      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 将会导致无法编译
        //      Incompatible types:
        //      Required: Uri
        //      Found: AsyncJob<Uri>   
            }
        });
        return savedUriAsyncJob;
    }

    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

哎。。。看起来没这么简单啊，下面修复返回的类型再试一次：

public class WeatherHelper {
    ApiWrapper apiWrapper;

    public AsyncJob<Uri> saveTheMaxDay(final String city) {
        //第一步,请求数据
        final AsyncJob<List<Weather>> weatherListAsyncJob = apiWrapper.queryWeathers(city);
        //第二步,处理数据
        final AsyncJob<Weather> maxDayAsyncJob = weatherListAsyncJob.map(new Function<List<Weather>, Weather>() {
            @Override
            public Weather call(List<Weather> weathers) {
                return findMaxDay(weathers);
            }
        });
        //第三步,存储数据

        AsyncJob<AsyncJob<Uri>> savedUriAsyncJob = maxDayAsyncJob.map(new Function<Weather, AsyncJob<Uri>>() {
            @Override
            public AsyncJob<Uri> call(Weather weather) {
                return apiWrapper.save(weather);
            }
        });
        return savedUriAsyncJob;
        //^^^^^^^^^^^^^^^^^^^^^^^ 将会导致无法编译
        //      Incompatible types:
        //      Required: AsyncJob<Uri>
        
    }

    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

这里我们只能拿到 AsyncJob<AsyncJob> 。看来还需要更进一步。我们需要压缩一层AsyncJob ，把两个异步操作当做一个单一的异步操作来对待。

现在我们需要一个参数为 AsyncJob 的 map 转换操作而不是 R。该操作类似于 map，但是该操作会把嵌套的 AsyncJob 压缩为（flatten ）一层 AsyncJob. 我们称之为 flatMap，实现如下：

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

    /**
     * 简单映射
     * @param func
     * @param <R> 输入类型
     * @return 输出类型
     */
    public <R> AsyncJob<R> map(final Function<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) {
                        R mapped = func.call(result);
                        callback.onResult(mapped);
                    }

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

    /**
     * 高级映射
     * @param func
     * @param <R>
     * @return
     */
    public <R> AsyncJob<R> flatMap(final Function<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(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);
                    }
                });
            }
        };
    }

}

看起来很多干扰代码，但是还好这些代码在客户端代码中并不会出现。现在我们的 WeathersHelper 如下：

public class WeatherHelper {
    ApiWrapper apiWrapper;

    public AsyncJob<Uri> saveTheMaxDay(final String city) {
        //第一步,请求数据
        final AsyncJob<List<Weather>> weatherListAsyncJob = apiWrapper.queryWeathers(city);
        //第二步,处理数据
        final AsyncJob<Weather> maxDayAsyncJob = weatherListAsyncJob.map(new Function<List<Weather>, Weather>() {
            @Override
            public Weather call(List<Weather> weathers) {
                return findMaxDay(weathers);
            }
        });
        //第三步,存储数据
        AsyncJob<Uri> savedUriAsyncJob = maxDayAsyncJob.flatMap(new Function<Weather, AsyncJob<Uri>>() {
            @Override
            public AsyncJob<Uri> call(Weather weather) {
                return apiWrapper.save(weather);
            }
        });
        
        return savedUriAsyncJob;
    }
    
    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

现在看起来简单多了，最终的代码看起来是不是有点眼熟啊？再仔细看看。还没发现？如果把匿名类修改为 Java 8 的 lambdas 表达式（逻辑是一样的，只是让代码看起来更清晰点）就很容易发现了。

public class WeatherHelper {
    ApiWrapper apiWrapper;

    public AsyncJob<Uri> saveTheMaxDay(final String city) {

        AsyncJob<List<Weather>> weatherListAsyncJob = apiWrapper.queryWeathers(city);
        AsyncJob<Weather> maxDayAsyncJob = weatherListAsyncJob.map(weathers -> findMaxDay(weathers));
        AsyncJob<Uri> savedUriAsyncJob = maxDayAsyncJob.flatMap(weather -> apiWrapper.save(weather));

        return savedUriAsyncJob;
    }

    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

这样看起来是不是就很清晰了。这个代码和刚刚开头的阻塞式代码是不是非常相似：

public class WeathersHelper {
    Api api;
    public Uri saveTheMaxDay(String city){
        List<Weather> weathers = api.queryWeathers(city);
        Weather max = findMaxDay(weathers);
        Uri savedUri = api.save(max);
        return savedUri;
    }
 
    private Weather findMaxDay(List<Weather> weathers) {
        return Collections.max(weathers);
    }
}

现在他们不仅逻辑是一样的，语义上也是一样的。太棒了！
同时我们还可以使用组合操作，现在把两个异步操作组合一起并返还另外一个异步操作。
异常处理也会传递到最终的回调接口中。

总结

下面来看看 RxJava 吧。
你没必要把上面代码应用到您的项目中去，这些简单的、线程不安全的代码只是 RxJava 的一部分。只有一些名字上的不同：
AsyncJob 等同于 Observable，不仅仅可以返回一个结果，还可以返回一系列的结果，当然也可能没有结果返回。
Callback 等同于 Observer，除了onNext(T t), onError(Throwable t)以外，还有一个onCompleted()函数，该函数在结束继续返回结果的时候通知Observable 。
abstract void start(Callback callback) 和 Subscription subscribe(final Observer observer) 类似，返回一个Subscription ，如果你不再需要后面的结果了，可以取消该任务。除了 map 和 flatMap 以外， Observable 还有很多其他常见的转换操作。
下一节,让我们认识一下RxJava 中关键的类。

最后编辑于：2018.05.21 15:59:45

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 203,098评论 5赞 476
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 85,213评论 2赞 380
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 149,960评论 0赞 336
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,519评论 1赞 273
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,512评论 5赞 364
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,533评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 37,914评论 3赞 395
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,574评论 0赞 256
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 40,804评论 1赞 296
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,563评论 2赞 319
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,644评论 1赞 329
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,350评论 4赞 318
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 38,933评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,908评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,146评论 1赞 259
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 42,847评论 2赞 349
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,361评论 2赞 342

RxJava前世今生

需求

同步方法实现

异步回调方法实现

代码优化

总结

推荐阅读更多精彩内容