我使用了Scala Futures 中TimeoutScheduler
介绍的- 内置超时?.
但是,现在我的程序不会像以前那样在没有TimeoutScheduler
.
我有两个Future
s:res1
和res2
. 两者都有 15 秒的超时。最后,我对两个s 进行排序,以便在回调Future
中正确关闭 HTTP 执行器。onComplete
不使用withTimeout
该程序后立即终止http.shutdown
。但是使用withTimeout
is 不会。为什么?一定还有未来……
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