Scala-ScheduledFuture
(7)
Akka tiene akka.pattern:
def after[T](duration: FiniteDuration, using: Scheduler)(value: ⇒ Future[T])(implicit ec: ExecutionContext): Future[T]
"Devuelve un scala.concurrent.Future que se completará con el éxito o el fracaso del valor proporcionado después de la duración especificada".
Estoy tratando de implementar el futuro programado en Scala. Me gustaría esperar un tiempo específico y luego ejecutar el cuerpo. Hasta ahora he intentado el siguiente, sencillo enfoque.
val d = 5.seconds.fromNow
val f = future {Await.ready(Promise().future, d.timeLeft); 1}
val res = Await.result(f, Duration.Inf)
Pero estoy obteniendo el TimeoutExcpetion sobre el futuro. ¿Es este incluso el enfoque correcto o simplemente debo usar el ScheduledExecutor de Java?
La solución más corta para esto, es probablemente usar scala-async:
import scala.async.Async.{async, await}
def delay[T](value: T, t: duration): Future[T] = async {
Thread.sleep(t.toMillis)
value
}
O en caso de que quiera retrasar la ejecución de un bloque.
def delay[T](t: duration)(block: => T): Future[T] async {
Thread.sleep(t.toMillis)
block()
}
Mi solución es bastante similar a la de Régis, pero uso Akka para programar:
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
}
No hay nada que hacer fuera de la caja usando solo la biblioteca estándar. Para los casos de uso más simples, puedes usar un pequeño ayudante como este:
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 ) )
}
}
Esto se puede utilizar de esta manera:
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits._
DelayedFuture( 5 seconds )( println("Hello") )
Tenga en cuenta que, a diferencia de los futuros programados de Java, esta implementación no le permitirá cancelar el futuro.
Podrías cambiar tu código a algo como esto:
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))
}
Pero no lo recomendaría. Al hacerlo, estarías bloqueando en un Futuro (bloqueando para esperar esa Promise que nunca se completará), y creo que el bloqueo en el ExecutionContext está muy desaconsejado. Me gustaría considerar el uso del ejecutor programado de java como dijiste o podrías usar Akka como se recomienda @ alex23.
Si desea programar la finalización sin Akka, puede usar un temporizador de Java regular para programar una promesa para completar:
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
}
Todas las demás soluciones utilizan akka o bloquean un subproceso por tarea retrasada. Una mejor solución (a menos que ya estés usando akka) es usar el ScheduledThreadPoolExecutor de java. Aquí hay un ejemplo de un envoltorio de scala para eso: