RxJava2.0 在安卓中的二级缓存策略

2,323 阅读8分钟

如有转载,请申明:
转载至 blog.csdn.net/qq_35064774…

前言

在上一篇 安卓网络数据缓存策略 中,介绍了安卓中数据的缓存策略,这篇将用RxJava2.0 实现 Json/Xml 数据的二级缓存。
对于 RxJava2.0 不了解的,可以看一下这篇入门教程 从零开始的RxJava2.0教程1-4

仿佛有一段时间没写博客了,吓得我都祭出了神图。

数据实时性高

为了便于没有看过上一篇教程的同学理解,我先把伪代码再贴一次。

如果 (存在缓存) {
    读取缓存并显示
}
请求网络
写入缓存
显示网络数据

上篇提到过,如果缓存可用,请求网络的时候,不应该显示正在加载的界面,网络请求失败的时候,也不应该显示错误界面。

为了优雅的实现这样一个多分支逻辑,我们需要用到 concat 操作符,和 1.x 中一样,将两个发射源按顺序连接成一个,这样先显示缓存,后显示网络数据的需求就完美的解决了。
不过需要注意的是,RxJava2.0 和 1.x 不一样, 所有的操作符都不能接收 null,所以,需要对缓存发射源和网络发射源进行一些额外的处理。

1. concat 连接缓存和网络数据

先给出最简单的代码,这段代码能实现基本功能,但在界面显示上会有一些逻辑问题,这个问题我们后面再解决。

Flowable.concat(localRepo.getHome(index), remoteRepo.getHome(index))
        .subscribe(new Consumer<AppListBean>() {
            @Override
            public void accept(AppListBean appListBean) throws Exception {
                getView().setData(appListBean);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                getView().showError(throwable, pullToRefresh);
            }
        });

可以看到,通过 concat 连接了本地和远程的数据源。成功或失败就通知界面显示。
我们在跟进 localReporemoteRepo 看一下如何处理 null 问题。

对于本地的源,是一个很简单的从文件读取数据,然后生成一个 Flowable,但需要注意的是,但文件不存在或数据有问题时,不能返回 null,相应的,我们返回一个空的发射源,也就是什么都不会发射的 Flowable。一个是避免 concat 收到 null 而抛出异常,另一个是方便后面逻辑判断。

public Flowable<AppListBean> getHome(@Query("index") int index) {
    return RxUtils.fromCache(cacheDir, "home" + index, AppListBean.class)
            .compose(RxUtils.<AppListBean>netScheduler());
}

// 从文件读取数据,并生成 Flowable
public static <T> Flowable<T> fromCache(final File dir, final String name, Class<T> type) {
    try {
        Gson gson = new Gson();
        T t = gson.fromJson(new FileReader(FileUtils.getJson(dir, name)), type);
        return Flowable.just(t);
    } catch (FileNotFoundException e) {
        e.printStackTrace();
    }
    return Flowable.empty();
}

对于远程的源,主要是对 retrofit 转换生成的 Flowable 进行出错拦截。

@Override
public Flowable<AppListBean> getHome(@Query("index") final int index) {
    return api.getHome(index)// retrofit转换得到的Flowable
            .compose(RxUtils.<AppListBean>netScheduler())// subscribeOn io observeOn mainThread
            .compose(RxUtils.<AppListBean>cache(FileUtils.getJson(cacheDir, "home" + index), 0 == index));// 缓存到本地,以及出错拦截。
}

// 缓存到本地,以及出错拦截
public static <T> FlowableTransformer<T, T> cache(final File file, final boolean isCache) {
    return new FlowableTransformer<T, T>() {
        @Override
        public Publisher<T> apply(Flowable<T> upstream) {
            return upstream.doOnNext(new Consumer<T>() {//获取数据成功时,缓存到本地
                @Override
                public void accept(T t) throws Exception {
                    if (isCache) {
                        Gson gson = new Gson();
                        String json = gson.toJson(t, t.getClass());
                        FileUtils.saveFileWithString(file, json);
                        Logger.d("cache success " + file);
                    }
                }
            }).onErrorResumeNext(new Function<Throwable, Publisher<? extends T>>() {// 出错拦截,当出现错误时,返回一个新的源而不是调用onError
                @Override
                public Publisher<? extends T> apply(Throwable throwable) throws Exception {
                    return Flowable.empty();// 这里返回一个空的发射源
                }
            });
        }
    };
}

这里我重点解释一下出错拦截,如果这里不调用 onErrorResumeNext 操作符,那么,当网络访问出错时,就会走 getView().showError(throwable, pullToRefresh); 这段 onError 逻辑,这样,只要网络出错,无论是否有本地缓存,界面都将显示一个错误,这显然不是我们要的,所以,这里对远程数据源进行出错拦截,一旦出错,就返回一个空的发射源。

2. 本地和远程都为空发射源的逻辑处理

上面那段代码似乎能起到我们想要的效果了,但一测试就会发现,当本地没有缓存,网络请求失败时,两者返回的都是空的发射源,也就是,界面既不显示数据,也不会显示出错。这显然不行,所以,我们还需要对 连接后的源进行非空检测。

Flowable.concat(localRepo.getHome(index), remoteRepo.getHome(index))
        .switchIfEmpty(new Flowable<AppListBean>() {// 空数据检测
            @Override
            protected void subscribeActual(Subscriber<? super AppListBean> s) {
                s.onError(new NoSuchElementException());
            }
        })
        .subscribe(new Consumer<AppListBean>() {
            @Override
            public void accept(AppListBean appListBean) throws Exception {
                getView().setData(appListBean);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                getView().showError(throwable, pullToRefresh);
            }
        });

这段代码与上段代码相比,只多了一个 switchIfEmpty 操作符,这个操作符的作用是,当发射源没有发送任何数据时,就会进入到该逻辑。在这个逻辑中,我们调用 s.onError(new NoSuchElementException()); 来进入到错误分支,这样就可以使界面显示出错信息了。

数据定期更新或不频繁变化

同样先把伪代码贴出来。

如果 (存在缓存 且 缓存未过期) {
    读取缓存并显示
    返回
}
请求网络
更新缓存
显示最新数据

有了上面一类缓存的基础,处理这个就容易多了。

Flowable.concat(localRepo.getHome(index), remoteRepo.getHome(index))
.firstOrError()// 最多发射一个数据,如果没有数据,则走 onError
.subscribe(new Consumer<AppListBean>() {
    @Override
    public void accept(AppListBean appListBean) throws Exception {
        getView().setData(appListBean);
    }
}, new Consumer<Throwable>() {
    @Override
    public void accept(Throwable throwable) throws Exception {
        getView().showError(throwable, pullToRefresh);
    }
});

与上面明显区别是, switchIfEmpty 换成了 firstOrError
注释上已经解释清楚了,这里详细介绍一下 localReporemoteRepo 里面的一些不同之处。

先看本地的发射源,与之前不同的是,多了一个 filter 操作符,这个是用过滤过期数据的,为了便于记录数据的过期时间,我在 bean 中加了一个 cacheTime 表示缓存的时间戳。

public Flowable<AppListBean> getHome(@Query("index") final int index) {
    return RxUtils.fromCache(cacheDir, "home" + index, AppListBean.class)
            .compose(RxUtils.<AppListBean>netScheduler())
            .filter(new Predicate<AppListBean>() {// 屏蔽过期数据
                @Override
                public boolean test(AppListBean appListBean) throws Exception {
                    if (appListBean.cacheTime + CACHE_TIME < System.currentTimeMillis()) {// 已经过期
                        // clean cache
                        RxUtils.cleanCache(FileUtils.getJson(cacheDir, "home" + index));
                        return false;
                    }
                    return true;
                }
            });
}

然后再看远程数据源,多了一个 doOnNext 操作符,这个是在写入缓存前,把当前的时间存到 bean 中去。

@Override
public Flowable<AppListBean> getHome(@Query("index") final int index) {
    return api.getHome(index)
            .doOnNext(new Consumer<AppListBean>() {
                @Override
                public void accept(AppListBean appListBean) throws Exception {
                    appListBean.cacheTime = System.currentTimeMillis();
                }
            })
            .compose(RxUtils.<AppListBean>netScheduler())
            .compose(RxUtils.<AppListBean>cache(FileUtils.getJson(cacheDir, "home" + index), 0 == index));
}

到此为止,RxJava2.0 的缓存实现已经介绍完了,这里给出的只是鄙人的一些见解,如果你有更好的方案,随时欢迎交流。

关于RxCache

RxCache 是一个很优秀的安卓数据缓存库,用在实际项目中,可以节省不少开发时间。我不推崇重复造轮子,但原理性的东西不能完全不知道,所以在最后给大家推荐这样一个库,希望能给你的开发带来帮助。

1. 不适合数据实时性高的策略

需要注意的是,RxCache 并不适合 数据实时性高 的缓存策略,因为它的加载机制如下:

请求数据(使用缓存) {
    如果 (缓存可用) {
        使用缓存数据
        返回  
    }
    请求网络数据
    存储缓存
}

请求数据(不使用缓存) {
    删除缓存
    请求网络数据
    存储缓存(如果请求失败,就没有缓存了)
}

所以,要实现请求本地数据后,再请求网络数据就不是那么容易。
当然也不是完全不行。经过我一下午的调试,最终整合出一个勉强可用的实现。

对于 Providers 的定义,返回的内容用 Reply 包裹,这样就可以知道返回的数据是来自网络还是缓存。

Observable<Reply<AppListBean>> getHome(Observable<AppListBean> home, DynamicKey index, EvictDynamicKey update);

使用缓存请求数据源,然后是利用 flatMap 中途修改发射源。
如果数据来自缓存,则给发射源连接一个网络请求的发射源。

public Observable<AppListBean> getHome(final int index) {
    Observable<Reply<AppListBean>> local = providers.getHome(api.getHome(index), new DynamicKey(index), new EvictDynamicKey(false))
            .flatMap(new Function<Reply<AppListBean>, ObservableSource<Reply<AppListBean>>>() {// 中途根据情况修改发射源
                @Override
                public ObservableSource<Reply<AppListBean>> apply(Reply<AppListBean> appListBeanReply) throws Exception {
                            Logger.d("get cache success");
                    Observable<Reply<AppListBean>> cache = Observable.just(appListBeanReply);
                    if (appListBeanReply.getSource() != Source.CLOUD// 数据来自缓存,则需要再加一个网络的请求
                                    && NetworkUtils.isAvailableByPing(BaseApplication.getContext())) {// 网络可用时才请求
                        // concat a remote request
                        Observable<Reply<AppListBean>> remote = providers.getHome(api.getHome(index), new DynamicKey(index), new EvictDynamicKey(true))
                                .onErrorResumeNext(new Function<Throwable, ObservableSource<? extends Reply<AppListBean>>>() {// 网络请求出错时,返回一个空发射源,而不是走onError
                                    @Override
                                    public ObservableSource<? extends Reply<AppListBean>> apply(Throwable throwable) throws Exception {
                                        return Observable.empty();
                                    }
                                });
                        return Observable.concat(cache, remote).distinct();
                    }
                    return cache;
                }
            });
    return local.map(new Function<Reply<AppListBean>, AppListBean>() {// 转换成数据
        @Override
        public AppListBean apply(Reply<AppListBean> appListBeanReply) throws Exception {
            return appListBeanReply.getData();
        }
    }).compose(RxUtils.<AppListBean>netScheduler());
}

这样处理之后,正常情况下,无论缓存是否可用,都会请求一次网络。这样就达到了我们想要的目的。
但有个特殊情况,当缓存可用时,我们附加了更新数据的请求,虽然网络已经被验证过可用,但并不能保证一定访问成功,一旦出错,我们就会失去缓存。因为在请求网络前,缓存就被删除了,而请求失败时,不会生成缓存。

但这种特殊情况很少出现,可能服务器异常,又或者网络请求还未完成时,突然没网了。
所以我称之为勉强可以接受的实现。

从这修改的工作量来看,自己实现缓存策略或许更加合适。

2. 适合数据定期更新或不频繁变化

对于这类缓存策略,RxCache支持的非常好,你可以通过注解设置过期时间,是否加密等。
你可以放心的调用不使用缓存的请求方法,当过期或者没有缓存的时候,会自动请求网络数据。
简直不要太舒服→_→