Scala Futures：非确定性输出

Question

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

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))

结果并不总是在成功和失败之间交替。在成功出现之前，可能会有不止几次失败的运行。

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

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线程模型是如何准确工作的——但是你明白我的意思。尽管如此，我还是得到了这个我无法掌握的非确定性输出with.So....我的问题是：这个非确定性输出来自哪里？

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

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))

虽然如果我减少重试次数（重试（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))

Answer 1

这最终对我起了作用:

(以下< code>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"))
    }
  }

不需要< code > atomic integer ——在我看来，这让事情变得更加复杂。需要一个< code >同步的包装器。

我必须强调的事实是，这只是出于演示的目的，在生产代码中使用这样的设计可能不是最好的主意(阻止对< code>calc方法的调用)。应该改用< code>recoverWith实现。

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

Answer 2

匿名用户

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

看起来具体问题是您正在测试重试，然后稍后更新它。所有四个线程都可能在更新值之前对其进行测试。在这种情况下，他们都将看到0并抛出错误。

解决方案是将< code>retries转换为< code>AtomicInteger并使用< code>getAndIncrement。这将自动检索该值并递增，因此每个线程都将看到适当的值。

更新以下评论：另一个答案已经解释了为什么同时启动多个线程，所以我不会在这里重复。当多个线程并行运行时，日志记录的顺序始终是不确定的。

Answer 3

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

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

f.fallbackTo(Future(calc()))

是的。以及它与

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

有两个重要的区别:

>

在回退到的情况下，立即创建未来（calc（）），因此（几乎）立即开始执行calc（）。因此，原始未来和回退未来同时运行。在使用恢复的情况下，只有在原始未来失败后才创建回退未来。这种差异会影响日志记录顺序。此外，这意味着对var重试的访问是并发的，因此您可能会看到所有线程实际失败的情况，因为对retries的一些更新丢失。

另一个棘手的一点是（突出显示是我的）

创建一个新的未来，如果该未来已成功完成，则保留该未来的结果;如果没有，则保存该未来成功完成，则保存该未来的结果。如果两个期货都失败了，那么最终的未来将持有第一个未来的可抛出对象。

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

  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))

并且从来没有任何其他“失败”值。

请注意，在这里，我显式地通过了在重试时不命中争用条件的尝试。

更多评论的回答 （1月 28）

在我的前一个示例中，我显式传递＜code＞尝试＜/code＞的原因是，这是确保逻辑上第一个＜code＞calc＜/code＞创建的＜code＞IllegalArgumentException＜/code＞在所有（甚至不是非常现实的）线程调度下都将＜code＞1＜/code＞作为其值的最简单方法。

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

  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")
    }
  }

这样你就避免了经典的托克托问题。