I am trying to implement scheduled future in Scala. I would like it to wait specific time and then execute the body. So far I tried the following, simple approach
val d = 5.seconds.fromNow
val f = future {Await.ready(Promise().future, d.timeLeft); 1}
val res = Await.result(f, Duration.Inf)
but I am getting the TimeoutExcpetion on the future. Is this even the correct approach or should I simply use the ScheduledExecutor from Java?
Akka 有 akka.pattern：
def after[T](duration: FiniteDuration, using: Scheduler)(value: ⇒ Future[T])(implicit ec: ExecutionContext): Future[T]
“返回一个 scala.concurrent.Future，它将在指定的持续时间后以提供的值的成功或失败来完成。”
仅使用标准库就没有任何开箱即用的功能。对于大多数简单的用例，您可以使用一个小助手，例如：
object DelayedFuture {
import java.util.{Timer, TimerTask}
import java.util.Date
import scala.concurrent._
import scala.concurrent.duration.FiniteDuration
import scala.util.Try
private val timer = new Timer(true)
private def makeTask[T]( body: => T )( schedule: TimerTask => Unit )(implicit ctx: ExecutionContext): Future[T] = {
val prom = Promise[T]()
schedule(
new TimerTask{
def run() {
// IMPORTANT: The timer task just starts the execution on the passed
// ExecutionContext and is thus almost instantaneous (making it
// practical to use a single Timer - hence a single background thread).
ctx.execute(
new Runnable {
def run() {
prom.complete(Try(body))
}
}
)
}
}
)
prom.future
}
def apply[T]( delay: Long )( body: => T )(implicit ctx: ExecutionContext): Future[T] = {
makeTask( body )( timer.schedule( _, delay ) )
}
def apply[T]( date: Date )( body: => T )(implicit ctx: ExecutionContext): Future[T] = {
makeTask( body )( timer.schedule( _, date ) )
}
def apply[T]( delay: FiniteDuration )( body: => T )(implicit ctx: ExecutionContext): Future[T] = {
makeTask( body )( timer.schedule( _, delay.toMillis ) )
}
}
这可以像这样使用：
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits._
DelayedFuture( 5 seconds )( println("Hello") )
请注意，与 java 预定期货不同，此实现不会让您取消未来。
如果你想在没有 Akka 的情况下安排完成，你可以使用常规的 Java Timer 来安排 Promise 的完成：
def delay[T](delay: Long)(block: => T): Future[T] = {
val promise = Promise[T]()
val t = new Timer()
t.schedule(new TimerTask {
override def run(): Unit = {
promise.complete(Try(block))
}
}, delay)
promise.future
}
我的解决方案与 Régis 的解决方案非常相似，但我使用 Akka 来安排：
def delayedFuture[T](delay: FiniteDuration)(block: => T)(implicit executor : ExecutionContext): Future[T] = {
val promise = Promise[T]
Akka.system.scheduler.scheduleOnce(delay) {
try {
val result = block
promise.complete(Success(result))
} catch {
case t: Throwable => promise.failure(t)
}
}
promise.future
}
您可以将代码更改为以下内容：
val d = 5.seconds.fromNow
val f = Future {delay(d); 1}
val res = Await.result(f, Duration.Inf)
def delay(dur:Deadline) = {
Try(Await.ready(Promise().future, dur.timeLeft))
}
但我不会推荐它。这样做，你会在 Future 中阻塞（阻塞以等待Promise永远不会完成的那个），我认为阻塞ExecutionContext是非常不鼓励的。我会按照您所说的那样考虑使用 java 调度执行程序，或者您可以按照@alex23 的建议考虑使用 Akka。
所有其他解决方案都使用 akka 或阻止每个延迟任务的线程。更好的解决方案（除非您已经在使用 akka）是使用 java 的 ScheduledThreadPoolExecutor。这是一个 scala 包装器的示例：
Shortest solution for this, is probably making use of scala-async:
import scala.async.Async.{async, await}
def delay[T](value: T, t: duration): Future[T] = async {
Thread.sleep(t.toMillis)
value
}
Or in case you want delayed execution of a block
def delay[T](t: duration)(block: => T): Future[T] async {
Thread.sleep(t.toMillis)
block()
}