IO 上的 RxJava 竞争条件答案

【问题标题】：RxJava race condition on IOIO 上的 RxJava 竞争条件
【发布时间】：2017-10-17 15:51:34
【问题描述】：

问题

当其中一个 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()

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

【问题讨论】：

我不认为 rx 可以帮助你进行同步，除非你愿意在调用 1 之后发生调用 2 和 3。SharedPrefs 应该是线程安全的，但它有一个不同的锁阅读和写作。由于您正在执行写入->读取，我建议您使用自己的锁将注销代码包装在某种 LoginManager 中。
我建议重新考虑您的实施。如果您在使用 Rx 时遇到竞争条件，那么您使用 Rx 不正确。尝试退后一步，重新考虑您要完成的工作。
@WoogieNoogie 问题不在 RxJava 中。问题是 RxJava 进行的异步调用（正确的和期望的）都与未同步的应用程序的一部分通信，因此所有这些都具有过时的信息。

标签： android rx-java rx-android

【解决方案1】：

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

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

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

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

关键是重构各个 API 调用的状态，将关于何时何地进行注销的决策推到更高的层次。

【讨论】：

好吧，情况并非如此，我没有检查 api 调用的范围，因为所有检查都发生在调用本身的身份验证部分内。仅对于此示例，我有 3 个调用，但应用程序的某些部分有 2 个、4 个或只有一个。结果代码必须更加灵活。谢谢你的帮助。

【解决方案2】：

我设法以非常简单的方式解决了我的问题。由于Authenticator 仅在 -

时触发

来自远程 Web 服务器或代理服务器的身份验证挑战

发生，因此我可以同步负责刷新令牌的代码部分。这样，每当发生刷新令牌失败时，logout() 方法将被同步并更新每个调用中的数据。

代码

override fun authenticate(route: Route, response: Response): Request? {
    synchronized(this, {
      ###refresh token code###
    })
}

【讨论】：