在 RxJava 中调用 Observable 的任何订阅者回调之前执行操作答案

【问题标题】：Do an action before any subscriber callback of an Observable is called in RxJava在 RxJava 中调用 Observable 的任何订阅者回调之前执行操作
【发布时间】：2018-03-08 17:41:45
【问题描述】：

我有一个Observable 链（RxJava 1），我想在可观察对象完成其工作时（就在它发出某些东西之前）执行一些一次性操作，但 在任何订阅者回调之前调用（我想更新一些AtomicBoolean 值，并让所有订阅者看到新值）。

以线程安全的方式执行此操作的最佳选择是什么？

我检查了doOnTerminate，doOnCompleted，doAfterTerminate 但似乎无法保证订单：我放了一些日志，有时第一个订阅者回调会在任何一个之前执行那些方法。

有什么方法可以明确地说“我希望在任何回调之前执行此操作”，还是我只需要在return从我的实用程序方法中订阅它之前作为Observable的第一个订阅者，并且更新该订阅者中的AtomicBoolean？

【问题讨论】：

标签： java rx-java rx-android

【解决方案1】：

您可以在返回给订阅者的观察者链中使用doOnNext() 运算符。

AtomicBoolean firstCome = new AtomicBoolean( false );
...
Observable<SomeData> observerChain =
  emitSomeData()
    .doOnNext( ignored -> firstCome.set( true ) )
    .share();
return observerChain;

编辑：使用share() 运算符允许多个订阅者。

【讨论】：

publish() 返回一个 ConnectableObservable 你丢失了哪个属性，并且链永远不会是 connect()-ed 这种方式。
实际上，我误解了我遇到的问题，而且基本上不是（仅）在onNext 中使用我的布尔值，我还必须添加一个onCompleted 回调 - 见下文。

【解决方案2】：

回答我自己的问题：这段代码有点复杂，不是我的，所以问的时候没看懂。

基本上代码如下； observable 发出 1 或 2 个值。该代码依赖于订阅者的onNext 回调。大多数时候（令人惊讶）它工作得很好，即onNext 会看到原始作者希望的布尔值，但有时onAfterTerminate 只会在所有onNext 回调之后执行。

     private Observable<...> getData() {
        return dataStore.getData()
              .doOnNext(...)
              .map(...)
              .doOnSubscribe(() -> myBoolean = true)
              .doAfterTerminate(() -> myBoolean = false)
              .map(...)
      }

    Subscription mySub = getData()
            .observeOn(AndroidSchedulers.mainThread(), true)
            .subscribe(onNextDoSomethingWithABoolean,
               onErrorDoSomethingWithABoolean);

主要的解决方法是将onCompleted 回调添加到订阅并在那里使用布尔值：

    Subscription mySub = getData()
            .observeOn(AndroidSchedulers.mainThread(), true)
            .subscribe(onNextDoSomethingWithABoolean,
              onErrorDoSomethingWithABoolean,
              onCompletedDoSomethingWithTheBoolean);

除此之外我还：

从boolean 更改为AtomicBoolean
将doAfterTerminate 更改为doOnTerminate

无论如何，这段代码可能都可以重写以完全避免可变状态。

【讨论】：