Scala Futures：非确定性输出答案

【问题标题】：Scala Futures: Non deterministic outputScala Futures：非确定性输出
【发布时间】：2019-06-20 15:58:44
【问题描述】：

我是 Scala 的新手，我正在通过创建一些重试方案来练习 Futures 库。这样做我得到了以下代码：

import scala.concurrent.{Await, Future}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

object Retries extends App {

  var retries = 0

  def resetRetries(): Unit = retries = 0

  def calc() = if (retries > 3) 10 else {
    retries += 1
    println(s"I am thread ${Thread.currentThread().getId} This is going to fail. Retry count $retries")
    throw new IllegalArgumentException("This failed")
  }

  def fCalc(): Future[Int] = Future(calc())

  resetRetries()

  val ff = fCalc() // 0 - should fail
    .fallbackTo(fCalc()) // 1 - should fail
    .fallbackTo(fCalc()) // 2 - should fail
    .fallbackTo(fCalc()) // 3 - should fail
    .fallbackTo(fCalc()) // 4 - should be a success

  Await.ready(ff, 10.second)

  println(ff.isCompleted)
  println(ff.value)
}

每次我运行这段代码都会得到不同的结果。我得到的结果示例如下

输出 1

I am thread 12 This is going to fail. Retry count 1
I am thread 14 This is going to fail. Retry count 3
I am thread 13 This is going to fail. Retry count 2
I am thread 11 This is going to fail. Retry count 1
I am thread 12 This is going to fail. Retry count 4
true
Some(Failure(java.lang.IllegalArgumentException: This failed))

输出 2

I am thread 12 This is going to fail. Retry count 2
I am thread 11 This is going to fail. Retry count 1
I am thread 13 This is going to fail. Retry count 3
I am thread 14 This is going to fail. Retry count 4
true
Some(Success(10))

输出 3

I am thread 12 This is going to fail. Retry count 1
I am thread 11 This is going to fail. Retry count 1
I am thread 12 This is going to fail. Retry count 2
I am thread 12 This is going to fail. Retry count 3
I am thread 12 This is going to fail. Retry count 4
true
Some(Failure(java.lang.IllegalArgumentException: This failed))

结果并非总是在成功和失败之间交替出现。可能会出现多次失败的运行，直到出现成功的运行。

据我了解，“我是线程 x 这将失败。重试计数 x”的日志应该只有 4 条，这些应该如下：

I am thread a This is going to fail. Retry count 1
I am thread b This is going to fail. Retry count 2
I am thread c This is going to fail. Retry count 3
I am thread d This is going to fail. Retry count 4

不一定按这个顺序——因为我不知道 Scala 线程模型是如何工作的——但你明白我的意思。尽管如此，我得到了我无法处理的不确定性输出。所以...... 我的问题是：这种不确定的输出来自哪里？

我想提一下，以下重试机制始终会产生相同的结果：

import scala.concurrent.{Await, Future}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

object Retries extends App {

  var retries = 0

  def resetRetries(): Unit = retries = 0

  def calc() = if (retries > 3) 10 else {
    retries += 1
    println(s"I am thread ${Thread.currentThread().getId} This is going to fail. Retry count $retries")
    throw new IllegalArgumentException("This failed")
  }

  def retry[T](op: => T)(retries: Int): Future[T] = Future(op) recoverWith { case _ if retries > 0 => retry(op)(retries - 1) }

  resetRetries()
  val retriableFuture: Future[Future[Int]] = retry(calc())(5)
  Await.ready(retriableFuture, 10 second)

  println(retriableFuture.isCompleted)
  println(retriableFuture.value)
}

输出

I am thread 11 This is going to fail. Retry count 1
I am thread 12 This is going to fail. Retry count 2
I am thread 11 This is going to fail. Retry count 3
I am thread 12 This is going to fail. Retry count 4
true
Some(Success(10))

如果我减少重试次数 (retry(calc())(3))，结果是预期的失败未来

I am thread 11 This is going to fail. Retry count 1
I am thread 12 This is going to fail. Retry count 2
I am thread 11 This is going to fail. Retry count 3
I am thread 12 This is going to fail. Retry count 4
true
Some(Failure(java.lang.IllegalArgumentException: This failed))

【问题讨论】：

标签： scala future

【解决方案1】：

虽然从技术上讲@Tim 是正确的，但我认为他并没有真正回答这个问题。

我相信你困惑的真正根源是你对结构的误解：

f.fallbackTo(Future(calc()))

确实如此。以及它与

的不同之处

f.recoverWith({ case _ => Future(calc())})

有两个重要的区别：

在fallbackTo 的情况下，Future(calc()) 会立即创建，因此（几乎）会立即开始执行calc()。因此，原始未来和备用未来同时运行。在recoverWith 的情况下，后备未来仅在原始未来失败后创建。这种差异会影响记录顺序。这也意味着对var retries 的访问是并发的，因此您可能会看到所有线程实际上都失败的情况，因为对retries 的一些更新丢失了。
另一个棘手的问题是fallbackTo 是documented（突出显示是我的）

创建一个新的未来，如果它成功完成则保存这个未来的结果，或者如果没有成功完成，则保存那个未来的结果。 如果两个future都失败了，resulting future持有第一个future的可投掷对象。

这种差异不会真正影响您的示例，因为您在所有失败尝试中抛出的异常都是相同的，但如果它们不同，它可能会影响结果。例如，如果您将代码修改为：

  def calc(attempt: Int) = if (retries > 3) 10 else {
    retries += 1
    println(s"I am thread ${Thread.currentThread().getId} This is going to fail. Retry count $retries")
    throw new IllegalArgumentException(s"This failed $attempt")
  }

  def fCalc(attempt: Int): Future[Int] = Future(calc(attempt))

  val ff = fCalc(1) // 0 - should fail
      .fallbackTo(fCalc(2)) // 1 - should fail
      .fallbackTo(fCalc(3)) // 2 - should fail
      .fallbackTo(fCalc(4)) // 3 - should fail
      .fallbackTo(fCalc(5)) // 4 - should be a success

那么你应该得到这两个结果中的任何一个

Some(Failure(java.lang.IllegalArgumentException: This failed 1))
Some(Success(10))

从来没有任何其他“失败”的价值。

请注意，这里我明确传递了attempt 以不达到retries 的竞争条件。

回答更多 cmets（1 月 28 日）

我在前面的示例中显式传递attempt 的原因是，它是确保由逻辑上第一个calc 创建的IllegalArgumentException 将得到1 作为其值的最简单方法（即使不是非常现实）线程调度。

如果您只想让所有日志具有不同的值，有一个更简单的方法：使用局部变量！

  def calc() = {
    val retries = atomicRetries.getAndIncrement()
    if (retries > 3) 10 
    else {
      println(s"I am thread ${Thread.currentThread().getId} This is going to fail. Retry count $retries")
      throw new IllegalArgumentException(s"This failed $retries")
    }
  }

这样可以避免经典的TOCTOU 问题。

【讨论】：

引入一个 AtomicInteger 并没有解决日志记录的问题，但它确实解决了成功输出的问题，根据你所说的不应该解决，因为：“为了你的例子返回Some(Success(10)) 最后一个线程应该赢得与所有其他线程的比赛，并且在所有其他线程之后开始”
@Niko，对不起，我错了。我不知道我当时在想什么。这种说法显然是错误的。一个成功的最后未来将导致一个成功的总结果，而与时间无关。我仍然觉得fallbackTo 和recoverWith 之间的区别很有趣，但这与这个问题无关。我会尽快修正我的答案。
至于日志记录，我不确定您到底看到了什么问题。您可以选择何时开始“后备”期货：与主要期货同时或按顺序启动。在我能想到的任何真正的“重试”场景中，您希望它们是连续的，否则它不是重试。 fallbackTo 不保证这个顺序，相反它同时运行Futures，因此日志记录是同时发生的。您仍然可以通过使用一些额外的同步来确保日志记录的顺序，但这是实现recoverWith 的简单事情的奇怪方法
@Tim , @SergGr 我的担心与日志的 order 无关。输出为： A (...) 重试计数 2 (...) 重试计数 2 (...) 重试计数 3 (...) 重试计数 4 B (...) 重试计数 2 (... ) 重试计数 1 (...) 重试计数 3 (...) 重试计数 4 2 在第一种情况下打印两次，我使用 AtomicInteger 表示 times 变量。这是我没有得到的。即使线程同时启动（在fallbackTo 的情况下），据我了解，如果times 是AtomicInteger，则不会发生这种情况。
@Niko，我相信我的回答的要点可能会在这些讨论中丢失，要点是如果您使用fallbackTo，您得到的逻辑不能真正称为“重试” " 因为所有的尝试都是同时运行的。对于“重试”，您确实需要使用 recoverWith 仅在前一次失败时运行下一次尝试。

【解决方案2】：

这不是 Scala 问题，而是更一般的多线程问题，值为 retries。您有多个线程在没有任何同步的情况下读取和写入此值，因此您无法预测每个线程何时运行或它将看到什么值。

看起来具体问题是您正在测试retries，然后再更新它。有可能所有四个线程都在它们中的任何一个更新它之前测试该值。在这种情况下，他们都会看到 0 并抛出错误。

解决方案是将retries 变成AtomicInteger 并使用getAndIncrement。这将自动检索值并递增它，因此每个线程都会看到适当的值。

更新以下cmets：另一个答案已经解释了为什么会同时启动多个线程，这里不再赘述。由于多个线程并行运行，日志记录的顺序总是不确定的。

【讨论】：

【解决方案3】：

这最终对我有用：

（calc() 方法的以下代码充分解决了有关记录重复和期货的不确定性结果的问题）

var time = 0
  var resetTries = time = 0

  def calc() = this.synchronized {
    if (time > 3) 10 else {
      time += 1
      println(s"I am thread ${Thread.currentThread().getId} This is going to fail. Retry count $time") // For debugging purposes
      throw new IllegalStateException(("not yet"))
    }
  }

不需要AtomicInteger - 在我看来让事情变得更加复杂。需要 synchronised 包装器。

我必须强调一个事实，这只是为了演示目的，在生产代码中使用这样的设计可能不是最好的主意（阻止对 calc 方法的调用）。应该改用recoverWith 实现。

感谢@SergGr、@Tim 和@MichalPolitowksi 的帮助

【讨论】：