java - 如何在RxJava中实现链锁

Question

我想在我的应用程序中实现锁定，以便同时只允许一个链片段执行，而其他链片段则相互等待。

例如:

    val demoDao = DemoDao() // data that must be accessed only by one rx-chain fragment at one time

    Observable.range(0, 150)
            .subscribeOn(Schedulers.io())
            .flatMapCompletable {
                dataLockManager.lock("action") { // fragment-start
                    demoDao.get()
                        .flatMapCompletable { data ->
                            demoDao.set(...)
                        }
                }                                // fragment-end
            }
            .subscribe()

    Observable.range(0, 100)
            .subscribeOn(Schedulers.io())
            .flatMapCompletable {
                dataLockManager.lock("action") { // fragment-start
                    demoDao.get()
                            .flatMapCompletable { data ->
                                demoDao.set(...)
                            }
                }                                // fragment-end
            }
            .subscribe()

我尝试通过自定义 Completable.create 和 CountDownLatch 来实现它，但这可能会导致死锁。

在这一点上我很糟糕。你能给我推荐什么吗？

Answer 1

要序列化对 demoDao.get() 的访问，有几种方法可以实现这一点，但尽量不要使用锁来实现这一点，因为这可能会导致 react 流中出现死锁初学者(正如您所发现的)。

如果您确实想使用锁，则应确保流信号上没有持有任何锁，例如向下游的发射或向上游的请求。在这种情况下，您可以使用锁(短暂的)。

一种方法是将两个流的操作合并为一个(例如merge)，并在该一个流上执行demoDao 操作。

另一种方法是使用 PublishSubject.create().serialized() 创建一个 PublisheSubject，它可以执行 demoDao.get() 的工作下游并仅订阅一次。那么您提到的两个来源可以.doOnNext(x -> subject.onNext())。取决于每个源是否必须独立了解故障，或者 PublishSubject 订阅是通知故障的唯一位置是否可以接受。