android - IO上的RxJava竞争条件

问题

由于缺少授权令牌，其中一个API调用最终失败时，我正在尝试注销用户。但是我无法将Schedulers.io()上的操作与SharedPrefs.commit()同步。所有呼叫都在同一时间进行，因此所有呼叫都尝试同时注销。
我要达到的目标是只有第一个触发注销，剩下两个要注意，这是发生的，根本不触发任何事件。

施工

我有3个API调用，这些调用会通过令牌自动实现。
令牌保存在SharedPrefs中。
所有API调用均在Schedulers.io()上进行。
调用令牌刷新失败时，我正在尝试使用注销方法从SharedPrefs中删除​​数据。

下面是注销方法代码：

 Flowable.fromCallable { isUserLogged() }
            .filter { it }
            .flatMap { logoutUseCase.execute() }
            .doOnComplete { showLogin() }
            .subscribeOn(Schedulers.single())
            .blockingSubscribe()

在连接到Authenticator的OkHttpClient类中使用Retrofit调用此方法

最佳答案

您可以使用Observable.amb()运算符解决比赛。

如果在没有身份验证令牌的情况下每个API调用均发出错误，则可以组织它们：

Observable.amb( apiCall1Observable, apiCall2Observable, apiCall3Observable )
  .observeOn(Schedulers.io())
  .doOnError( error -> {
    logoutUseCase.execute()
   } )
  .subscribe( value -> {
   },
   error -> {
   });

amb()运算符将选择第一个来发出值或终止事件。这将触发仅执行一次的logoutUseCase.execute()。

关键是从各个API调用中重构状态，从而将有关何时何地注销的决定推向更高的层次。