使用 Retrofit 和 Realm 纠正 RxJava 中的流程

我正在结合 RxJava 和 Retrofit 来实现网络 API，并使用 Realm 作为数据库。我几乎可以正常工作，但我想知道这是否是正确的方法和事件流程。所以，这里是RetrofitApiManager.

public class RetrofitApiManager {

    private static final String BASE_URL = "***";

    private final ShopApi shopApi;

    public RetrofitApiManager(OkHttpClient okHttpClient) {

        // GSON INITIALIZATION

        Retrofit retrofit = new Retrofit.Builder()
                .client(okHttpClient)
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                .addConverterFactory(GsonConverterFactory.create(gson))
                .baseUrl(BASE_URL)
                .build();

        shopApi = retrofit.create(ShopApi.class);
    }

    public Observable<RealmResults<Shop>> getShops() {
        return shopApi.getShops()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(response -> {
                    Realm realm = Realm.getDefaultInstance();
                    realm.executeTransaction(realm1 -> 
                            realm1.copyToRealmOrUpdate(response.shops));
                    realm.close();
                })
                .flatMap(response -> {
                    Realm realm = Realm.getDefaultInstance();
                    Observable<RealmResults<Shop>> results = realm.where(Shop.class)
                            .findAllAsync()
                            .asObservable()
                            .filter(RealmResults::isLoaded);
                    realm.close();
                    return results;
                });
    }
}

这是获取的电话RealmResults<Shop>里面一个Fragment.

realm.where(Shop.class)
        .findAllAsync()
        .asObservable()
        .filter(RealmResults::isLoaded)
        .first()
        .flatMap(shops -> 
                shops.isEmpty() ? retrofitApiManager.getShops() : Observable.just(shops))
        .subscribe(
                shops -> initRecyclerView(),
                throwable -> processError(throwable));

这是我的问题：

1. 这是像上面的示例中那样的连锁事件的正确方法还是我应该以不同的方式管理它们？

2. 可以使用吗Realm实例在getShops()方法并在那里关闭我，或者将它作为参数传递然后以某种方式管理它会更好吗？虽然，这个想法对于线程和调用来说似乎有点问题Realm.close()总是在正确的时间。

1）我会尝试在后台线程上做尽可能多的事情，现在你正在 UI 线程上做很多工作。

2)

  public Observable<RealmResults<Shop>> getShops() {
        return shopApi.getShops()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(response -> {
                    try(Realm realm = Realm.getDefaultInstance()) {
                        realm.executeTransaction(realm1 -> 
                            realm1.insertOrUpdate(response.shops));
                    } // auto-close
                })
                .flatMap(response -> {
                    try(Realm realm = Realm.getDefaultInstance()) {
                        Observable<RealmResults<Shop>> results = realm.where(Shop.class)
                            .findAllAsync()
                            .asObservable()
                            .filter(RealmResults::isLoaded);
                    } // auto-close
                    return results;
                });
    }

所有 Realm 数据都是延迟加载的，因此仅在 Realm 实例打开时可用，因此在检索它后关闭它很可能不起作用。在你的情况下，虽然你在主线程上进行平面映射，所以很可能那里已经有一个打开的实例。

如果你愿意你可以使用copyFromRealm()获取可以跨线程移动并且不再连接到 Realm 的非托管数据，但它们也会失去实时更新功能并占用更多内存。

它可能会这样做：

  public Observable<RealmResults<Shop>> getShops() {
        return shopApi.getShops()
                .subscribeOn(Schedulers.io())
                .doOnNext(response -> {
                    try(Realm realm = Realm.getDefaultInstance()) {
                        realm.executeTransaction(realm1 -> 
                            realm1.copyToRealmOrUpdate(response.shops));
                    } // auto-close
                })
                .observeOn(AndroidSchedulers.mainThread())
                .flatMap(response -> {
                    Observable<RealmResults<Shop>> results = realm.where(Shop.class)
                            .findAllAsync()
                            .asObservable()
                            .filter(RealmResults::isLoaded);
                    return results;
                });

或者，您可以将网络请求视为副作用，只依赖领域在发生更改时通知您（IMO 更好的方法，因为您将网络与数据库访问分开，这就是存储库模式的含义）

public Observable<RealmResults<Shop>> getShops() {
    // Realm will automatically notify this observable whenever data is saved from the network
    return realm.where(Shop.class).findAllAsync().asObservable()
            .filter(RealmResults::isLoaded)
            .doOnNext(results -> {
                if (results.size() == 0) {
                    loadShopsFromNetwork();
                }
            }); 
}

private void loadShopsFromNetwork() {
    shopApi.getShops()
            .subscribeOn(Schedulers.io())
            .subscribe(response -> {
                try(Realm realm = Realm.getDefaultInstance()) {
                    realm.executeTransaction(r -> r.insertOrUpdate(response.shops));
                } // auto-close
            });
}