基于 RxJava 的响应式编程

Loading Data

• loading-data-from-multiple-sources-with-rxjava

假设我需要从网络上获取一些数据。每次需要数据的时候，我都可以简单的访问网络，但是，将数据缓存到磁盘或内存则可以更有效率。

更明确的说，我希望是这样的： 1、偶尔的从网络上获取新数据。 2、然而可以尽快的恢复数据(通过缓存网络数据的结果)。

按优先级加载有效数据

给每一数据源(网络、磁盘和内存)一个Observable接口，我们可以通过两个操作：concat()和first()，来实现一个简单的解决方案。concat()持有多个Observables，并且把它们连接在队列里。first()仅从队列里中获取到第一个条目。因此，如果你使用concat().first()可以从多个数据源中获取到第一个。

// Our sources (left as an exercise for the reader)
Observable<Data> memory = ...;
Observable<Data> disk = ...;
Observable<Data> network = ...;

// Retrieve the first source with data
Observable<Data> source = Observable
  .concat(memory, disk, network)
  .first();

这种模式的关键是 concat()只在需要资源的时候才会订阅每个子 Observable。如果数据被缓存，就不需要通过速度慢的数据源来获取数据。注意 concat()中 Observables 数据源的顺序问题，因为它们是被一个接一个检索出来的。在实际应用场景中，我们还需要在first中判断获取到的数据是否有效以及是否过期，只要进行简单的修正即可：

Observable<Data> source = Observable.concat(
        sources.memory(),
        sources.disk(),
        sources.network()
)
.first(data -> data != null && data.isUpToDate());

自动保存数据

很显然，下一步就是保存数据源。如果，你没有将网络请求的结果保存到磁盘，将磁盘的地址保存在内存中，那就再也没法挽救啦！上面所有的代码就是让网咯请求持久化。我的解决方式是在每次发出请求的时候保存或缓存数据源：

Observable<Data> networkWithSave = network.doOnNext(data -> {
  saveToDisk(data);
  cacheInMemory(data);
});

Observable<Data> diskWithCache = disk.doOnNext(data -> {
  cacheInMemory(data);
});

现在，如果你使用networkWithSave和diskWithCache，数据都将会在你下载的时候自动保存。(这种策略的另外一个好处就是networkWithSave/diskWithCache可以在任何地方使用，不仅仅在我们的多个数据源模式下。)

日志记录

有时候，我们还需要记录下每次请求的命中情况，譬如有时候我们需要去测试下缓存的命中率，可以用 compose 方法来实现：

// Save network responses to disk and cache in memory
return observable.doOnNext(data -> {
    disk = data;
    memory = data;
})
        .compose(logSource("NETWORK"));
...
// Simple logging to let us know what each source is returning
Observable.Transformer<Data, Data> logSource(final String source) {
    return dataObservable -> dataObservable.doOnNext(data -> {
        if (data == null) {
            System.out.println(source + " does not have any data.");
        } else if (!data.isUpToDate()) {
            System.out.println(source + " has stale data.");
        } else {
            System.out.println(source + " has the data you are looking for!");
        }
    });
}

完整的测试代码可以参考这里