社交类应用多数据源管理求助（Room+Remote+RxJava）

嘿，我来帮你解决这个多数据源的问题！你的需求其实就是典型的「先显示本地缓存，同时拉取远程更新」的社交APP模式，RxJava完全可以轻松搞定，咱们一步步来：

首先，修正Repository的返回类型 #

你现在的getMovies()返回List<Movie>是同步的，完全不符合RxJava异步流的设计，必须改成返回Observable<Resource<List<Movie>>>（Observable更适合处理多发射场景）。这里我们用一个Resource包装类来区分本地和远程数据，方便UI层精准处理。

编写Resource包装类（可选但推荐） #

这个类帮我们标记数据来源，让UI明确知道什么时候显示本地数据，什么时候处理远程更新：

public class Resource<T> {
    public enum DataSource { LOCAL, REMOTE }
    public final DataSource source;
    public final T data;

    public Resource(DataSource source, T data) {
        this.source = source;
        this.data = data;
    }
}

实现Repository的核心逻辑 #

这里我们用Observable.merge()来并行发起本地和远程请求——本地数据因为是磁盘查询会先返回，远程请求在后台同时进行，完美契合你的需求：

public class MoviesRepository {
    private final RoomLocalDataSource roomLocalDataSource;
    private final RemoteDataSource remoteDataSource;

    // 构造函数注入数据源
    public MoviesRepository(RoomLocalDataSource roomLocalDataSource, RemoteDataSource remoteDataSource) {
        this.roomLocalDataSource = roomLocalDataSource;
        this.remoteDataSource = remoteDataSource;
    }

    public Observable<Resource<List<Movie>>> getMovies() {
        // 1. 本地数据源流：获取本地数据并标记为LOCAL
        Observable<Resource<List<Movie>>> localStream = roomLocalDataSource.getMoviesFromLocal()
                .toObservable()
                .map(localMovies -> new Resource<>(Resource.DataSource.LOCAL, localMovies))
                .onErrorReturn(throwable -> new Resource<>(Resource.DataSource.LOCAL, Collections.emptyList()));

        // 2. 远程数据源流：获取远程数据后先保存到本地，再标记为REMOTE
        Observable<Resource<List<Movie>>> remoteStream = remoteDataSource.getMoviesFromRemote()
                .toObservable()
                .doOnSuccess(remoteMovies -> roomLocalDataSource.saveMoviesToLocal(remoteMovies)) // 记得给RoomLocalDataSource添加保存方法
                .map(remoteMovies -> new Resource<>(Resource.DataSource.REMOTE, remoteMovies))
                .onErrorResumeNext(throwable -> Observable.empty()); // 远程失败不影响本地数据显示

        // 3. 合并两个流，并行执行
        return Observable.merge(localStream, remoteStream);
    }
}

ViewModel层如何处理 #

在ViewModel里订阅这个流，根据数据来源做不同处理：

public class MoviesViewModel extends ViewModel {
    private final MoviesRepository repository;
    public final MutableLiveData<List<Movie>> moviesLiveData = new MutableLiveData<>();
    public final MutableLiveData<Boolean> hasNewPosts = new MutableLiveData<>();

    public MoviesViewModel(MoviesRepository repository) {
        this.repository = repository;
        loadMovies();
    }

    private void loadMovies() {
        repository.getMovies()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(resource -> {
                    if (resource.source == Resource.DataSource.LOCAL) {
                        // 第一次收到本地数据，直接显示到RecyclerView
                        moviesLiveData.setValue(resource.data);
                        hasNewPosts.setValue(false);
                    } else if (resource.source == Resource.DataSource.REMOTE) {
                        // 远程数据拉取成功，显示「查看新帖子」按钮
                        hasNewPosts.setValue(true);
                    }
                }, throwable -> {
                    // 处理全局错误，比如Toast提示网络异常
                });
    }

    // 点击「查看新帖子」按钮时调用，刷新为最新本地数据（已包含远程拉取的内容）
    public void refreshMovies() {
        roomLocalDataSource.getMoviesFromLocal()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(moviesLiveData::setValue);
    }
}

为什么你之前用concat不行？ #

concat()是串行执行的：先完成本地请求，再发起远程请求，这会导致远程请求被延迟，完全不符合你「同时拉取远程」的需求。而merge()是并行执行两个流，本地数据先返回显示，远程请求在后台同步进行，正好匹配Instagram/LinkedIn的机制。

内容的提问来源于stack exchange，提问作者Mina Mikhail