Scala中超时的未来

Question

假设我有一个函数，它调用一个阻塞可中断操作。我想在超时的情况下异步运行它。也就是说，我想在超时到期时中断函数。

所以我正在尝试做这样的事情：

import scala.util.Try
import scala.concurrent.Future

def launch(f: () => Unit, timeout: Int): Future[Try[Unit]] = {

  val aref = new java.util.concurrent.atomic.AtomicReference[Thread]() 

  import ExecutionContext.Implicits.global
  Future {Thread.sleep(timeout); aref.get().interrupt} // 1
  Future {aref.set(Thread.currentThread); Try(f())}    // 2
}

问题是 （1） 中的 aref 可以为 null，因为 （2） 尚未将其设置为当前线程。在这种情况下，我想等到 aref 设置完毕。最好的方法是什么？

Answer 1

如果添加一个< code>CountDownLatch，就可以实现您想要的行为。(请注意，在很多很多的< code>Future中发生阻塞(即，在< code>await处停滞)可能会导致线程池的饥饿。)

import scala.util.Try
import scala.concurrent.Future

def launch(f: () => Unit, timeout: Int): Future[Try[Unit]] = {

  val aref = new java.util.concurrent.atomic.AtomicReference[Thread]()
  val cdl = new java.util.concurrent.CountDownLatch(1)

  import ExecutionContext.Implicits.global
  Future {Thread.sleep(timeout); cdl.await(); aref.get().interrupt}   // 1
  Future {aref.set(Thread.currentThread); cdl.countDown(); Try(f())}  // 2
}

Answer 2

我也需要相同的行为，所以这就是我解决它的方式。我基本上创建了一个对象，该对象创建一个计时器，如果将来未在指定的持续时间内完成，则使用TimeoutException使promise失败。

package mypackage

import scala.concurrent.{Promise, Future}
import scala.concurrent.duration.FiniteDuration
import akka.actor.ActorSystem
import scala.concurrent.ExecutionContext.Implicits.global

object TimeoutFuture {

  val actorSystem = ActorSystem("myActorSystem")
  def apply[A](timeout: FiniteDuration)(block: => A): Future[A] = {
    val promise = Promise[A]()
    actorSystem.scheduler.scheduleOnce(timeout) {
      promise tryFailure new java.util.concurrent.TimeoutException
    }

    Future {
      try {
        promise success block
      }
      catch {
        case e:Throwable => promise failure e
      } 
    }

    promise.future
  }
}

Answer 3

您可以使用Await实现稍微简单一点的方法。< code>Await.result方法将超时持续时间作为第二个参数，并在超时时引发< code>TimeoutException。

try {
  import scala.concurrent.duration._
  Await.result(aref, 10 seconds);
} catch {
    case e: TimeoutException => // whatever you want to do.
}