当前位置：首页 > news >正文

Android中使用RxJava实现网络请求与缓存策略

news 2025/8/21 5:40:16

前言

在现代Android开发中，网络请求是应用不可或缺的一部分，而良好的缓存策略可以显著提升用户体验。RxJava作为一种响应式编程框架，能够优雅地处理异步操作和事件流。本文将介绍如何结合RxJava实现网络请求与缓存策略。

一、RxJava基础回顾

RxJava的核心概念包括：

Observable：被观察者，数据源
Observer：观察者，接收数据
Operator：操作符，用于转换、过滤、组合数据流
Scheduler：调度器，控制线程切换

java

Observable.just("Hello RxJava").subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(text -> Log.d("RxJava", text));

二、网络请求实现

1. 使用Retrofit+RxJava进行网络请求

首先添加依赖：

gradle

implementation 'com.squareup.retrofit2:retrofit:2.9.0'
implementation 'com.squareup.retrofit2:adapter-rxjava3:2.9.0'
implementation 'com.squareup.retrofit2:converter-gson:2.9.0'

定义API接口：

java

public interface ApiService {@GET("user/{id}")Observable<User> getUser(@Path("id") String userId);
}

创建Retrofit实例：

java

Retrofit retrofit = new Retrofit.Builder().baseUrl("https://api.example.com/").addConverterFactory(GsonConverterFactory.create()).addCallAdapterFactory(RxJava3CallAdapterFactory.create()).build();ApiService apiService = retrofit.create(ApiService.class);

2. 发起网络请求

java

apiService.getUser("123").subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<User>() {@Overridepublic void onSubscribe(Disposable d) {// 可以在这里显示加载进度条}@Overridepublic void onNext(User user) {// 处理返回的用户数据}@Overridepublic void onError(Throwable e) {// 处理错误}@Overridepublic void onComplete() {// 请求完成}});

三、缓存策略实现

1. 内存缓存 + 磁盘缓存 + 网络请求

我们可以实现一个三级缓存策略：

首先检查内存缓存
如果没有，检查磁盘缓存
如果还是没有，发起网络请求
将网络请求结果缓存到内存和磁盘

2. 实现缓存管理器

java

public class CacheManager {private static final String TAG = "CacheManager";private static CacheManager instance;private LruCache<String, Object> memoryCache;private File cacheDir;private CacheManager(Context context) {// 初始化内存缓存 (这里设置为应用最大内存的1/8)final int maxMemory = (int) (Runtime.getRuntime().maxMemory() / 1024);final int cacheSize = maxMemory / 8;memoryCache = new LruCache<String, Object>(cacheSize) {@Overrideprotected int sizeOf(String key, Object value) {// 计算每个缓存对象的大小return value.toString().length() / 1024;}};// 初始化磁盘缓存目录cacheDir = new File(context.getCacheDir(), "network_cache");if (!cacheDir.exists()) {cacheDir.mkdirs();}}public static synchronized CacheManager getInstance(Context context) {if (instance == null) {instance = new CacheManager(context);}return instance;}// 内存缓存操作public synchronized void putToMemory(String key, Object value) {if (memoryCache.get(key) == null && value != null) {memoryCache.put(key, value);}}public synchronized Object getFromMemory(String key) {return memoryCache.get(key);}// 磁盘缓存操作public void putToDisk(String key, Object value) {File file = new File(cacheDir, key);try {ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(file));oos.writeObject(value);oos.close();} catch (IOException e) {e.printStackTrace();}}public Object getFromDisk(String key) {File file = new File(cacheDir, key);if (!file.exists()) return null;try {ObjectInputStream ois = new ObjectInputStream(new FileInputStream(file));Object value = ois.readObject();ois.close();return value;} catch (IOException | ClassNotFoundException e) {e.printStackTrace();return null;}}// 清除缓存public void clearMemoryCache() {memoryCache.evictAll();}public void clearDiskCache() {File[] files = cacheDir.listFiles();if (files != null) {for (File file : files) {file.delete();}}}
}

3. 结合RxJava实现缓存策略

java

public Observable<User> getUserWithCache(String userId) {return Observable.concat(getFromMemory(userId),getFromDisk(userId),getFromNetwork(userId)).filter(user -> user != null).firstElement().toObservable();
}private Observable<User> getFromMemory(String userId) {return Observable.create(emitter -> {User user = (User) CacheManager.getInstance(context).getFromMemory(userId);if (user != null) {emitter.onNext(user);}emitter.onComplete();});
}private Observable<User> getFromDisk(String userId) {return Observable.create(emitter -> {User user = (User) CacheManager.getInstance(context).getFromDisk(userId);if (user != null) {// 放入内存缓存CacheManager.getInstance(context).putToMemory(userId, user);emitter.onNext(user);}emitter.onComplete();}).subscribeOn(Schedulers.io());
}private Observable<User> getFromNetwork(String userId) {return apiService.getUser(userId).doOnNext(user -> {// 保存到内存和磁盘CacheManager.getInstance(context).putToMemory(userId, user);CacheManager.getInstance(context).putToDisk(userId, user);});
}

4. 使用示例

java

getUserWithCache("123").subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(user -> {// 更新UI}, throwable -> {// 处理错误});

四、高级缓存策略

1. 缓存过期策略

java

public class CacheItem<T> implements Serializable {private T data;private long timestamp;public CacheItem(T data) {this.data = data;this.timestamp = System.currentTimeMillis();}public boolean isExpired(long expiryTime) {return (System.currentTimeMillis() - timestamp) > expiryTime;}public T getData() {return data;}
}// 修改getFromDisk方法
public Observable<User> getFromDisk(String userId) {return Observable.create(emitter -> {CacheItem<User> cacheItem = (CacheItem<User>) CacheManager.getInstance(context).getFromDisk(userId);if (cacheItem != null) {if (!cacheItem.isExpired(TimeUnit.MINUTES.toMillis(10))) { // 10分钟过期User user = cacheItem.getData();CacheManager.getInstance(context).putToMemory(userId, user);emitter.onNext(user);} else {// 缓存过期，删除磁盘缓存CacheManager.getInstance(context).removeFromDisk(userId);}}emitter.onComplete();}).subscribeOn(Schedulers.io());
}

2. 先显示缓存再更新网络数据

java

public Observable<User> getUserWithCacheFirst(String userId) {Observable<User> memory = getFromMemory(userId);Observable<User> disk = getFromDisk(userId);Observable<User> network = getFromNetwork(userId).doOnNext(user -> {// 保存到内存和磁盘CacheManager.getInstance(context).putToMemory(userId, user);CacheManager.getInstance(context).putToDisk(userId, user);});return Observable.concat(memory, disk).filter(user -> user != null).take(1).concatWith(network).distinctUntilChanged();
}

五、错误处理与重试机制

java

getUserWithCache("123").retryWhen(throwableObservable -> throwableObservable.zipWith(Observable.range(1, 3), (throwable, retryCount) -> {if (retryCount < 3 && throwable instanceof IOException) {return retryCount;}throw Exceptions.propagate(throwable);}).flatMap(retryCount -> Observable.timer((long) Math.pow(2, retryCount), TimeUnit.SECONDS)).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(user -> {// 更新UI}, throwable -> {// 处理错误});