使用 RxJava2 中的一系列运算符一次处理一个 PublishSubject 发射答案

【问题标题】：Process one PublishSubject emission at a time with a chain of operators in RxJava2使用 RxJava2 中的一系列运算符一次处理一个 PublishSubject 发射
【发布时间】：2021-02-05 09:22:39
【问题描述】：

我有一个PublishSubject<Event>。

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

这是通过链接大致如下完成的：

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

subject 可能会在某些条件下发出两个快速的Events，假设间隔为 10 毫秒。

问题是 getCachedData()，第二次发射在 getCachedData() 第一次发射完成后立即关闭，即在 getCachedData() em>cleanCache() 有机会在第二次发射之前清理数据库。

我想以某种方式将这些 flatMaps 组合到一个观察者中，以便 subject 在整个链完成之前不会进行新的排放，最好没有任何手工制作的信号量.

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

我看到一些建议将 toSerialized() 添加到主题，但现在我认为它与链的工作方式无关。

我看到还有 lift() 和 compose() 运算符。我试图将所有 flatMaps 放在后者中，这并没有改变行为。前者我仍然想知道。

【问题讨论】：

标签： rx-java2 sequential publishsubject

【解决方案1】：

将它们放在concatMapX 子流中：

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

我看到一些建议将 toSerialized() 添加到主题，但现在我认为它与链的工作方式无关

除非您实际驱动由它返回并从多个线程返回的Subject，否则它对您的流程没有实际影响。

【讨论】：

太棒了！非常感谢。