将 Realm 与 RxJava 2 结合使用

我在 Android 应用程序中使用 RxJava 2，并集成 Realm。据我所知，Realm 默认只支持 RxJava 1，并且允许Observable查询时返回RealmResults<?>，像这样：

Realm.getDefaultInstance()
    .where(VideoBundle.class)
    .findAll()
    .asObservable()
    .first()

返回的 Observable 来自 RxJava 1。如何一起使用 Realm 和 RxJava 2？我遇到了2个相关问题，发现here https://github.com/realm/realm-java/issues/3497 and here https://github.com/realm/realm-java/issues/4135，但没有找到简洁的答案。此外，文档（在这里找到：https://realm.io/docs/java/latest/#rxjava https://realm.io/docs/java/latest/#rxjava）提到创建自定义RxObservableFactory，但没有提供有关如何执行此操作的资源。

Realm 如何与已经使用 RxJava 2 的项目一起使用？

解决方案是使用 Flowable 包装 RealmResults，并使用最新的背压策略。

private io.reactivex.Flowable<RealmResults<_>> getSomeItems() {
    return io.reactivex.Flowable.create(new FlowableOnSubscribe<RealmResults<__>>() {
        @Override
        public void subscribe(FlowableEmitter<RealmResults<__>> emitter)
                throws Exception {
            Realm realm = Realm.getDefaultInstance();
            RealmResults<__> results = realm.where(__.class).findAllSortedAsync("__");

            final RealmChangeListener<RealmResults<__>> listener = _realm -> {
                if(!emitter.isUnsubscribed() && results.isLoaded()) {
                     emitter.onNext(results);
                }
            };
            emitter.setDisposable(Disposables.fromRunnable(() -> {
                results.removeChangeListener(listener);
                realm.close();
            }));
            results.addChangeListener(listener);
        }
    }, BackpressureStrategy.LATEST)
    .subscribeOn(AndroidSchedulers.mainThread())
    .unsubscribeOn(AndroidSchedulers.mainThread());

从 Realm 4.0.0-RC1 及更高版本开始，我上面展示的这种行为是在使用中烘焙的realmResults.asFlowable().

Disposable subscription = realm.where(__.class)
                               .findAllSortedAsync("__")
                               .asFlowable()
                               .filter(RealmResults::isLoaded)
                               .subscribe(...);