java - 如何保证RXJava Observable指令在线程范围内执行？

Question

我目前将 RXJava 与数据库事务结合使用。这些数据库事务不是线程安全的，因此必须在单个线程范围内执行所有操作。

一个解决方案是将 Observable 转换为阻塞的 observable 并调用 single 或 last 方法。这样所有可观察的指令都在同一个线程中执行(如果没有另外指定)。一旦处理完所有指令，try-with-resources 语句将调用关闭。

尝试 (Trx tx = db.trx()) {
obs.toBlocking().single();
}


或者，调度程序可用于:

1. 将可观察指令绑定(bind)到某个线程
2. 在调用 onCompleted 后关闭事务。

是否有任何其他替代方法可用于确保可观察指令在单个线程范围内执行？

Answer 1

鉴于您的示例代码，我假设您是在可观察到的副作用(例如 doOnNext)中访问数据库，而不是在订阅调用中，不确定这是不是故意的，但您最好依靠订阅代码来解决副作用(对于写入)或您要合并的可观察源(对于读取)。

请注意，toBlocking().single() 不能保证您观察到的副作用会在当前线程中发生，但只能保证它们都已终止。

一个更简单的方法(对于写入)可能是这样做的:

obs.toList().subscribe(list -> 
    try (Trx tx = db.trx()) {
        // do DB stuff foreach item in list
    })

(如果您的 observable 没有终止，使用 buffer 而不是 toList 会很好)

如果你处理JDBC，你也可以看看rxjava-jdbc .实际上，即使您不这样做，库语法也可以为您提供有关如何将 DB 事务与 Rx 粘合的提示。