使用RxJava2中的一系列操作符一次处理一个PublishSubject排放

Question

我有一个发布主题

在每个新事件上，我都会触发一个数据库查询（一些本地缓存的数据），然后获取结果并尝试通过HTTP请求将其发布到服务器，当该服务器回复200时，我会进行另一个数据库查询以删除刚刚发送的行。

这大致是通过以下方式进行链接的：

subject
  .toSerialized()
  .flatMapMaybe { getCachedData() }
  .flatMap { uploadData() }
  .flatMapCompletable { cleanCache() }
  .subscribe()

在某些条件下，受试者可能会发出两个快速事件，比如说间隔10毫秒。

问题是，第二次发射的getCachedData（）在第一次发射的getCachedData（）完成后立即熄灭，即在cleanCache（）有机会在第二次发射之前清理数据库之前。

我希望以某种方式将这些平面图合并到一个观察者中，以便主体在整个链完成之前不会产生新的排放，最好没有任何手工信号量。

我在单线程池调度程序上执行subscribeOn（），它只对每个flatMap内的调用进行排序。

我看到了一些建议，可以添加到主题的序列化（），但现在我认为它与链的工作方式无关。

我也看到了升降（）和合成（）运算符。我试着把所有的平面图放在后者里面，这并没有改变行为。前者我还在想。




Answer 1

将它们放在concatMapX子流中：

subject
.concatMapCompletable {
    getCachedData()
    .flatMap { uploadData() }
    .flatMapCompletable { cleanCache() }
}

我看到了一些向主题添加toSerialized（）的建议，但现在我认为这与链的工作方式无关

它对您的流没有实际影响，除非您实际驱动它和从多个线程返回的主题。