我使用了在Scala Futures - built in timeout?中介绍的TimeoutScheduler
。
然而,现在我的程序没有TimeoutScheduler
之前那样终止。
我有两个Future
: res1
和res2
。它们都有15秒的超时时间。最后,我按顺序对这两个Future
进行合并,以便在onComplete
回调中正确关闭HTTP执行器。如果不使用withTimeout
,程序将在http.shutdown
之后立即终止。但是,如果使用withTimeout
,则不会。为什么?可能还有其他的future...
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://dev59.com/m2Qo5IYBdhLWcg3wI8VV
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
}
}
}
任何建议都受欢迎,祝好,/nm
HashedWheelTimer
不是守护进程(由Viktor Klang建议)? - nemronHashedWheelTimer
线程是非守护进程的,因此如果您想将该类用作高效的计时器机制,则必须显式停止它。虽然可能有不同的选项可用于计时器类,但您也可以研究一下。我使用HashedWheelTimer
只是为了完整性而已。如果您不喜欢明确关闭它,请不要感到必须使用它。 - cmbaxterHashedWheelTimer
似乎支持一个构造函数,它接受一个ThreadFactory
,用于生成表示计时器的线程。如果您使用该构造函数,应该能够传入自定义的ThreadFactory
,将返回的Thread
上的守护标志设置为true。 - cmbaxternew HashedWheelTimer(new DefaultDaemonicThreadFactory, 10, TimeUnit.MILLISECONDS)
,其中DefaultDaemonicThreadFactory
是Executors#DefaultThreadFactory
的一个原始副本,但具有t.setDaemon(true)
。它有效了!在这种情况下,TimeoutScheduler.timer.stop()
不是必要的。 - nemron