android - IO 上的 RxJava 竞争条件

Question

问题

当其中一个 API 调用由于缺少授权 token 而以失败告终时，我正在尝试注销我的用户。但我无法设法将 Schedulers.io() 上的操作与 SharedPrefs.commit() 同步。所有调用都是同时进行的，因此所有调用都同时尝试注销。 我想要实现的是只有第一个触发注销，剩下的两个注意到，它发生了，根本不触发任何东西。

建筑

我有 3 个 API 调用，它们使用 token 自动调整。 token 保存在 SharedPrefs 中。 所有 API 调用都是在 Schedulers.io() 上进行的。 当调用 token 刷新失败时，我正在尝试使用注销方法从我的 SharedPrefs 中删除数据。

注销方法代码如下:

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

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

Answer 1

您可以使用 Observable.amb() 运算符来解决竞争。

如果在没有身份验证 token 时每个 API 调用都发出错误，那么您可以这样组织它们:

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

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

关键是重构您个人 API 调用的状态，将有关何时何地注销的决策推向更高级别。