4

我使用了Scala Futures 中TimeoutScheduler介绍的- 内置超时？.

但是，现在我的程序不会像以前那样在没有TimeoutScheduler.

我有两个Futures:res1和res2. 两者都有 15 秒的超时。最后，我对两个s 进行排序，以便在回调Future中正确关闭 HTTP 执行器。onComplete不使用withTimeout该程序后立即终止http.shutdown。但是使用withTimeoutis 不会。为什么？一定还有未来……

import java.net.URI
import scala.util.{ Try, Failure, Success }
import dispatch._
import org.json4s._
import org.json4s.native.JsonMethods._
import com.typesafe.scalalogging.slf4j.Logging

object Main extends App with Logging {
  import scala.concurrent.ExecutionContext.Implicits.global

  val http   = Http
  val github = host("api.github.com").secure 

  import timeout._
  import scala.concurrent.duration._
  import scala.language.postfixOps
  import scala.collection.JavaConverters._

  val req: dispatch.Req =  github / "users" / "defunkt"
  val res1 = http(req > dispatch.as.Response(_.getHeaders().get("Server").asScala)) withTimeout (15 seconds) recover { case x => 
    logger.debug("Error1: " + x.toString) 
    Nil
  }
  val res2 = http(req > dispatch.as.Response(_.getHeaders().get("Vary").asScala)) withTimeout (15 seconds) recover { case x => 
    logger.debug("Error2: " + x.toString) 
    Nil
  }

  Future.sequence(res1 :: res2 :: Nil) onComplete { case _ => 
    http.shutdown()               // without using `withTimeout` the program terminated after `http.shutdow`
    TimeoutScheduler.timer.stop() // thanks to @cmbaxter
  }
}

object timeout {
  import java.util.concurrent.TimeUnit
  import scala.concurrent.Promise
  import scala.concurrent.duration.Duration
  import org.jboss.netty.util.Timeout
  import org.jboss.netty.util.TimerTask
  import org.jboss.netty.util.HashedWheelTimer
  import org.jboss.netty.handler.timeout.TimeoutException

   // cf. https://stackoverflow.com/questions/16304471/scala-futures-built-in-timeout
  object TimeoutScheduler {
    val timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS)
    def scheduleTimeout(promise: Promise[_], after: Duration) = {
      timer.newTimeout(new TimerTask{
        def run(timeout: Timeout){              
          promise.failure(new TimeoutException("Operation timed out after " + after.toMillis + " millis"))        
        }
      }, after.toNanos, TimeUnit.NANOSECONDS)
    }
  }
  implicit class FutureWithTimeout[T](f: Future[T]) {
    import scala.concurrent.ExecutionContext

    def withTimeout(after: Duration)(implicit ec: ExecutionContext) = {
      val prom = Promise[T]()
      val timeout = TimeoutScheduler.scheduleTimeout(prom, after)
      val combinedFut = Future.firstCompletedOf(List(f, prom.future))
      f onComplete { case result => timeout.cancel() }
      combinedFut
    }
  } 
}

欢迎任何建议，Best，/nm

4

1 回答 1

4

如果您完全按照描述使用我的代码，那么我的猜测是HashedWheelTimer位于TimeoutScheduler对象下方的 Netty 没有被终止。在这样调用之后，您可以尝试显式调用stop它http.shutdown：

TimeoutScheduler.timer.stop

现在，如果您希望 NettyHashedWheelTimer使用守护线程，那么您可以使用其中一个接受 a 的构造函数（我在 3.6.6-Final 中看到它们），ThreadFactory然后使用ThreadFactory将守护进程标志设置为 true的自定义.

于 2014-02-24T12:59:05.867 回答