关于Future.firstCompletedOf和Garbage Collect机制

Question

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

简单地说,我正在使用Future.firstCompletedOf从2 Future秒获得结果,两者都没有共享的东西而且彼此不关心.即使这是我想要解决的问题,垃圾收集器也无法回收第一个Result对象,直到两个Future完成.

所以我真的很好奇这背后的机制.有人可以从较低的层面解释它,或提供一些提示让我调查.

谢谢!

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.保存此函数的确切位置无关紧要,我们只知道它必须保存在某处,以便稍后可以在线程可用时在线程池中执行它.

关闭此功能completeFirst将关闭completeFirst.因此,只要还有一个未来(从p)等待完成,就会有一个参考,以futures防止它被垃圾收集(即使到那时机会p已经返回,firstCompletedOf从堆栈中删除).

当第一个未来完成时,它会将结果保存到承诺中(通过调用p).因为promise p.tryComplete保留了结果,所以结果至少与remhablle一样长p,并且正如我们所看到p的那样,只要至少有一个future未完成就可以到达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
}

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