领域：从异步事务创建 Observable答案

【问题标题】：Realm: Creating Observable from asynchronous Transaction领域：从异步事务创建 Observable
【发布时间】：2016-11-23 09:27:26
【问题描述】：

在我的 Android 应用程序中，我使用 Realm 和 RxJava，我想通过传递一个新值来更新一个对象。我为此使用realm.executeTransactionAsync()。但我想订阅此事件以在我使用此方法的类中进行错误处理。到目前为止我所做的是：

public Observable updateDogAge(final String dogId, final int age) {
    return Observable.create(new Observable.OnSubscribe<Dog>() {
        @Override
        public void call(final Subscriber<? super Dog> subscriber) {
            realm.executeTransactionAsync(realm1 -> realm1.where(Dog.class)
                            .equalTo(DOG_ID_FIELD, dogId)
                            .findFirst()
                            .setAge(age),
                    subscriber::onCompleted,
                    subscriber::onError);
        }
    });
}

现在我想获得更新此对象是否成功的布尔标志。现在我觉得只是打电话给subscriber::onCompleted 不是正确的解决方案。我也不确定Observable.onCreate()的使用。

你能告诉我从异步 Realm 事务中获取 Observable（或 Single）的正确方法是什么吗？

【问题讨论】：

标签： android asynchronous realm rx-java

【解决方案1】：

我建议为 RxJava 支持创建自己的“异步”事务。

public Observable updateDogAge(final String dogId, final int age) {
    return Observable.defer(() -> {
        try(Realm r = Realm.getDefaultInstance()) {
            r.executeTransaction((realm) -> realm.where(Dog.class)
                        .equalTo(DOG_ID_FIELD, dogId)
                        .findFirst()
                        .setAge(age));
            return Observable.just(true);
        } catch(Throwable e) {
            return Observable.error(e);
        }
    });
}

注意：我的 RxJava 有点朦胧，我不经常使用。

当然，您需要在后台线程上运行的调度程序上订阅它。

updateDogAge(id, age).subscribeOn(Schedulers.io())./*...*/

其实我觉得你可以改用Observable.fromCallable()：

public Observable updateDogAge(final String dogId, final int age) {
    return Observable.fromCallable(() -> {
        try(Realm r = Realm.getDefaultInstance()) {
            r.executeTransaction((realm) -> realm.where(Dog.class)
                        .equalTo(DOG_ID_FIELD, dogId)
                        .findFirst()
                        .setAge(age));
            return true;
        }
    });
}

【讨论】：

这看起来很简单，但是 try-catch 并没有捕获事务中的 Throwable。有什么建议吗？
catching the throwable in the transaction 是什么意思？ executeTransaction() 已经取消了异常事务，这就是它的用途。
我试图将它与executeTransactionAsync 一起使用。使用同步事务它可以工作。
不过有一件事，Realm 现在给我的信息是：REALM: Mixing asynchronous queries with local writes should be avoided. Realm will convert any async queries to synchronous in order to remain consistent. Use asynchronous writes instead. 我应该忽略它吗？
不，您应该在调用时将.subscibeOn(Schedulers.io()) 应用到updateDogAge

【解决方案2】：

我为 RxJava2 创建了 Completable

fun dbTransaction(db: Realm, transaction: Realm.Transaction): Completable {
    return Completable.create { emitter ->
        db.executeTransactionAsync(transaction, Realm.Transaction.OnSuccess { emitter.onComplete() }, Realm.Transaction.OnError { emitter.onError(it) })
    }
}

【讨论】：

【解决方案3】：

如果操作成功，则需要true，如果发生错误，则需要false，对吗？尝试始终返回 true 并添加 .onnErrorReturn，返回 false。您可以使用 .subscribeOn() 和 .observeOn() 方法和同步领域事务自己进行异步事务。但是，这样做，您每次都需要获取和释放 Realm 实例，因为它是线程相关的。这会更容易一些，因为您将根据操作成功显式返回 true 和 false。

【讨论】：

谢谢，听起来不错。你能给我看一些代码示例吗？
这就是下面的答案，包裹在 .subscribeOn(Schedulers.io).observeOn(AndroidSchedulers.mainThread())
并带有 .onErrorReturn(false) 运算符

【解决方案4】：

您还可以执行以下操作，以使用 executeTransactionAsync 而不是普通的 executeTransaction：

return Observable.fromEmitter(emitter -> {
    try (Realm realm = Realm.getDefaultInstance()) {
        realm.executeTransactionAsync(bgRealm -> {
            bgRealm.where(Dog.class)
                   .equalTo(DOG_ID_FIELD, dogId)
                   .findFirst()
                   .setAge(age));
            emitter.onNext(true);
        });
    } catch (Exception e) {
        emitter.onError(e);
    }
}, Emitter.BackpressureMode.BUFFER);

这样，您的所有 Realm 事务都将是异步的，您将摆脱所有关于“将异步查询与本地写入混合...”的警告。。

这种方法的一个可能缺点是 fromEmitter 仅适用于 RxJava 1.x 的更高版本，并且目前仍处于试验阶段（RxJava 1.2.3）。不确定 RxJava 2 是否有等效的方法。

【讨论】：