关于 Future.firstCompletedOf 和垃圾收集机制

Question

我在实际项目中遇到过这个问题，并通过我的测试代码和分析器证明了这一点。我没有粘贴“tl;dr”代码，而是向您展示一张图片，然后对其进行描述。

简单地说，我使用Future.firstCompletedOf 从 2 个Futures 中得到一个结果，这两个都没有共享的东西，也不关心彼此。尽管如此，这是我要解决的问题，垃圾收集器无法回收第一个 Result 对象，直到两个 Futures 完成。

所以我真的很好奇这背后的机制。有人可以从较低的层次解释它，或者提供一些提示让我研究一下。

谢谢！

PS：是不是因为他们共享同一个ExecutionContext？

**更新**按要求粘贴测试代码

object Main extends App{
  println("Test start")

  val timeout = 30000

  trait Result {
    val id: Int
    val str = "I'm short"
  }
  class BigObject(val id: Int) extends Result{
    override val str = "really big str"
  }

  def guardian = Future({
    Thread.sleep(timeout)
    new Result { val id = 99999 }
  })

  def worker(i: Int) = Future({
    Thread.sleep(100)
    new BigObject(i)
  })

  for (i <- Range(1, 1000)){
    println("round " + i)
    Thread.sleep(20)
    Future.firstCompletedOf(Seq(
      guardian,
      worker(i)
    )).map( r => println("result" + r.id))
  }

  while (true){
    Thread.sleep(2000)
  }
}

Answer 1

让我们看看firstCompletedOf是如何实现的：

def firstCompletedOf[T](futures: TraversableOnce[Future[T]])(implicit executor: ExecutionContext): Future[T] = {
  val p = Promise[T]()
  val completeFirst: Try[T] => Unit = p tryComplete _
  futures foreach { _ onComplete completeFirst }
  p.future
}

在执行{ futures foreach { _ onComplete completeFirst } 时，函数completeFirst 保存在某处 通过ExecutionContext.execute。这个函数到底保存在哪里无关紧要，我们只知道它必须保存在某个地方 以便稍后可以在线程可用时在线程池中选择并执行它。只有当 future 完成时，才不再需要对 completeFirst 的引用。

因为completeFirst 关闭了p，只要还有一个未来（来自futures）等待完成，就有一个对p 的引用可以防止它被垃圾收集（即使通过那一点很可能firstCompletedOf已经返回，从堆栈中删除p。

当第一个future 完成时，它会将结果保存到promise 中（通过调用p.tryComplete）。 因为promise p 持有结果，所以至少只要p 可达，结果就可达，并且正如我们所见，只要futures 的至少一个未来尚未完成，p 就可达。 这就是为什么在所有期货都完成之前无法收集结果的原因。

更新： 现在的问题是：它可以修复吗？我认为可以。我们所要做的就是确保第一个未来以线程安全的方式完成对 p 的引用，这可以通过使用 AtomicReference 的示例来完成。像这样的：

def firstCompletedOf[T](futures: TraversableOnce[Future[T]])(implicit executor: ExecutionContext): Future[T] = {
  val p = Promise[T]()
  val pref = new java.util.concurrent.atomic.AtomicReference(p)
  val completeFirst: Try[T] => Unit = { result: Try[T] =>
    val promise = pref.getAndSet(null)
    if (promise != null) {
      promise.tryComplete(result)
    }
  }
  futures foreach { _ onComplete completeFirst }
  p.future
}

我已经对其进行了测试，并且正如预期的那样，它确实允许在第一个未来完成后立即对结果进行垃圾收集。它应该在所有其他方面表现相同。