在RxJava中调用Observable的任何订户回调之前执行操作

Question

我有一个< code>Observable链(RxJava 1)，我想在Observable完成它的工作时(就在它发出某个东西之前)，但在调用任何订户回调之前(我想更新某个< code>AtomicBoolean值，并让所有订户都看到新值)，执行一些一次性操作。

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

我检查了 doOnOnTerioned，doOnComped，doTerTerminate，但似乎没有关于顺序的保证：我放了一些日志，有时在任何这些方法之前执行第一个订阅者回调。

有没有什么方法可以明确地说“我希望它在任何回调之前执行”，或者我是否必须在从我的实用方法< code >返回它之前，将自己订阅为< code>Observable的第一个订阅者，并在该订阅者中更新< code>AtomicBoolean？

Answer 1

回答我自己的问题：这段代码有点复杂，不是我的，所以我在问这个问题时没有很好地理解它。

基本上代码如下所示；可观察的发出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);

主要解决方法是将已完成回调添加到订阅中，并在那里使用布尔值：

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

除此之外，我还：

• 从布尔值更改为原子布尔值
• 已更改 do“终止”到“完成”

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

Answer 2

您可以在返回给订阅者的观察者链中使用< code>doOnNext()操作符。

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

编辑：使用＜code＞share（）