期货有一个方面,我不完全理解从官方教程参考http://docs.scala-lang.org/overviews/core/futures.html
scala的期货是否有某种内置的超时机制?让我们假设下面的例子是一个5gb的文本文件…"隐含"的隐含范围。全局"最终导致onFailure以非阻塞的方式触发,或者可以定义吗?如果没有某种默认超时,这是否意味着成功和失败都不可能发生?
import scala.concurrent._
import ExecutionContext.Implicits.global
val firstOccurence: Future[Int] = future {
val source = scala.io.Source.fromFile("myText.txt")
source.toSeq.indexOfSlice("myKeyword")
}
firstOccurence onSuccess {
case idx => println("The keyword first appears at position: " + idx)
}
firstOccurence onFailure {
case t => println("Could not process file: " + t.getMessage)
}
当您使用阻塞来获取Future
的结果时,您只会获得超时行为。如果您想使用非阻塞回调onComplete
、onSuccess
或onFailure
,那么您必须滚动自己的超时处理。Akka为参与者之间的请求/响应(?
)消息传递内置了超时处理,但不确定是否要开始使用Akka。FWIW,在Akka中,对于超时处理,它们通过Future.firstCompletedOf
组成两个Futures
,一个代表实际的异步任务,一个代表超时。如果超时计时器(通过HashedWheelTimer
)先弹出,则异步回调失败。
一个非常简单的滚动自己的例子可能是这样的。首先,一个用于调度超时的对象:
import org.jboss.netty.util.{HashedWheelTimer, TimerTask, Timeout}
import java.util.concurrent.TimeUnit
import scala.concurrent.duration.Duration
import scala.concurrent.Promise
import java.util.concurrent.TimeoutException
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)
}
}
然后是一个函数,用于获取Future并为其添加超时行为:
import scala.concurrent.{Future, ExecutionContext, Promise}
import scala.concurrent.duration.Duration
def withTimeout[T](fut:Future[T])(implicit ec:ExecutionContext, after:Duration) = {
val prom = Promise[T]()
val timeout = TimeoutScheduler.scheduleTimeout(prom, after)
val combinedFut = Future.firstCompletedOf(List(fut, prom.future))
fut onComplete{case result => timeout.cancel()}
combinedFut
}
请注意,我在这里使用的HashedWheelTimer
来自Netty。
所有这些答案都需要额外的依赖项。我决定使用java.util.Timer编写一个版本,这是将来运行函数的有效方法,在本例中是触发超时。
博客文章在这里有更多的细节
使用Scala的Promise,我们可以创建一个Future和timeout,如下所示:
package justinhj.concurrency
import java.util.concurrent.TimeoutException
import java.util.{Timer, TimerTask}
import scala.concurrent.duration.FiniteDuration
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.language.postfixOps
object FutureUtil {
// All Future's that use futureWithTimeout will use the same Timer object
// it is thread safe and scales to thousands of active timers
// The true parameter ensures that timeout timers are daemon threads and do not stop
// the program from shutting down
val timer: Timer = new Timer(true)
/**
* Returns the result of the provided future within the given time or a timeout exception, whichever is first
* This uses Java Timer which runs a single thread to handle all futureWithTimeouts and does not block like a
* Thread.sleep would
* @param future Caller passes a future to execute
* @param timeout Time before we return a Timeout exception instead of future's outcome
* @return Future[T]
*/
def futureWithTimeout[T](future : Future[T], timeout : FiniteDuration)(implicit ec: ExecutionContext): Future[T] = {
// Promise will be fulfilled with either the callers Future or the timer task if it times out
val p = Promise[T]
// and a Timer task to handle timing out
val timerTask = new TimerTask() {
def run() : Unit = {
p.tryFailure(new TimeoutException())
}
}
// Set the timeout to check in the future
timer.schedule(timerTask, timeout.toMillis)
future.map {
a =>
if(p.trySuccess(a)) {
timerTask.cancel()
}
}
.recover {
case e: Exception =>
if(p.tryFailure(e)) {
timerTask.cancel()
}
}
p.future
}
}
我刚刚为同事创建了一个TimeoutFuture
类:
TimeoutFuture
package model
import scala.concurrent._
import scala.concurrent.duration._
import play.libs.Akka
import play.api.libs.concurrent.Execution.Implicits._
object TimeoutFuture {
def apply[A](timeout: FiniteDuration)(block: => A): Future[A] = {
val prom = promise[A]
// timeout logic
Akka.system.scheduler.scheduleOnce(timeout) {
prom tryFailure new java.util.concurrent.TimeoutException
}
// business logic
Future {
prom success block
}
prom.future
}
}
使用val future = TimeoutFuture(10 seconds) {
// do stuff here
}
future onComplete {
case Success(stuff) => // use "stuff"
case Failure(exception) => // catch exception (either TimeoutException or an exception inside the given block)
}
指出:
- 假定玩!框架(但很容易适应)
- 每段代码都运行在同一个
ExecutionContext
中,这可能不是理想的。
Play框架包含Promise。超时,这样您可以编写如下代码
private def get(): Future[Option[Boolean]] = {
val timeoutFuture = Promise.timeout(None, Duration("1s"))
val mayBeHaveData = Future{
// do something
Some(true)
}
// if timeout occurred then None will be result of method
Future.firstCompletedOf(List(mayBeHaveData, timeoutFuture))
}
我很惊讶这不是Scala的标准。我的版本很短,没有依赖
import scala.concurrent.Future
sealed class TimeoutException extends RuntimeException
object FutureTimeout {
import scala.concurrent.ExecutionContext.Implicits.global
implicit class FutureTimeoutLike[T](f: Future[T]) {
def withTimeout(ms: Long): Future[T] = Future.firstCompletedOf(List(f, Future {
Thread.sleep(ms)
throw new TimeoutException
}))
lazy val withTimeout: Future[T] = withTimeout(2000) // default 2s timeout
}
}
使用示例import FutureTimeout._
Future { /* do smth */ } withTimeout
如果你想让写程序(承诺持有者)控制超时逻辑,使用akka.pattern。
val timeout = akka.pattern.after(10 seconds, system.scheduler)(Future.failed(new TimeoutException(s"timed out during...")))
Future.firstCompletedOf(Seq(promiseRef.future, timeout))
这样,如果你的承诺完成逻辑从来没有发生过,你的调用者的未来仍然会在某个点以失败完成。
可以指定等待未来时的超时时间:
对于scala.concurrent.Future
, result
方法允许您指定超时。
对于scala.actors.Future
, Futures.awaitAll
允许您指定超时。
我不认为有一个内置的超时执行一个Future。
还没有人提到akka-streams
。流有一个简单的completionTimeout
方法,将其应用于单源流就像Future一样。
但是,akka-streams也做了取消,所以它实际上可以结束源的运行,即它向源发出超时信号。
Monix Task
支持超时
import monix.execution.Scheduler.Implicits.global
import monix.eval._
import scala.concurrent.duration._
import scala.concurrent.TimeoutException
val source = Task("Hello!").delayExecution(10.seconds)
// Triggers error if the source does not complete in 3 seconds after runOnComplete
val timedOut = source.timeout(3.seconds)
timedOut.runOnComplete(r => println(r))
//=> Failure(TimeoutException)
此版本无需使用任何外部计时器(仅wait.result)
import scala.concurrent._
import scala.concurrent.duration.FiniteDuration
object TimeoutFuture {
def apply[A](
timeout: FiniteDuration
)(block: => A)(implicit executor: ExecutionContext): Future[A] =
try {
Future { Await.result(Future { block }, timeout) }
} catch {
case _: TimeoutException => Future.failed(new TimeoutException(s"Timed out after ${timeout.toString}"))
}
}
我正在使用这个版本(基于上面的Play示例),它使用Akka系统调度程序:
object TimeoutFuture {
def apply[A](system: ActorSystem, timeout: FiniteDuration)(block: => A): Future[A] = {
implicit val executionContext = system.dispatcher
val prom = Promise[A]
// timeout logic
system.scheduler.scheduleOnce(timeout) {
prom tryFailure new java.util.concurrent.TimeoutException
}
// business logic
Future {
try {
prom success block
} catch {
case t: Throwable => prom tryFailure t
}
}
prom.future
}
}
在Future IMO上指定超时最简单的方法是scala的内置机制,使用scala.concurrent.Await.ready
。如果Future花费的时间超过指定的超时,则会抛出TimeoutException
。否则,它将返回未来本身。下面是一个简单的人工示例
import scala.concurrent.ExecutionContext.Implicits._
import scala.concurrent.duration._
val f1: Future[Int] = Future {
Thread.sleep(1100)
5
}
val fDoesntTimeout: Future[Int] = Await.ready(f1, 2000 milliseconds)
val f: Future[Int] = Future {
Thread.sleep(1100)
5
}
val fTimesOut: Future[Int] = Await.ready(f, 100 milliseconds)
您可以使用Await
等待一个future结束。
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
val meaningOfLife: Int = Await.result(Future(42), 1.nano)
println (meaningOfLife)
上面打印42
您可能需要一个隐式的ExecutionContext
可用,在这种情况下,只需添加:
import scala.concurrent.ExecutionContext.Implicits.global
另一种方法是使用monix中的
Coeval
。这种方法并不适用于所有情况,您可以在这里阅读有关它的所有内容。其基本思想是,有时future并不真正占用任何时间,而是返回同步函数调用或值的结果,因此可以在当前线程上对该future进行评估。这对于测试和模拟未来也很有用。你也不必指定超时,这是预期的,但仍然很高兴不必担心这个。
你首先将未来转变为Task
,并将该任务包装在Coeval
中,然后交叉手指,等待看到你得到什么。这是一个非常简单的例子来展示它是如何工作的:
你需要一个隐式的Scheduler
才能使用它:
import monix.execution.Scheduler.Implicits.global
Coeval(Task.fromFuture(Future (42)).runSyncStep).value() match {
case Right(v) => println(v)
case Left(task) => println("Task did not finish")
}
以上操作完成并将42
打印到控制台。
Coeval(Task.fromFuture(Future {
scala.concurrent.blocking {
42
}
}).runSyncStep).value() match {
case Right(v) => println(v)
case Left(task) => println("Task did not finish")
}
这个例子打印Task did not finish
:
You can simply run the future to completion without giving any timeout interval by setting the timeout to infinite as below:
**import scala.concurrent.duration._
Await.result(run(executionContext), Duration.Inf)**
run function can be as below :
def run(implicit ec: ExecutionContext) = {
val list = Seq(
Future { println("start 1"); Thread.sleep(1000); println("stop 1")},
Future { println("start 2"); Thread.sleep(2000); println("stop 2")},
Future { println("start 3"); Thread.sleep(3000); println("stop 3")},
Future { println("start 4"); Thread.sleep(4000); println("stop 4")},
Future { println("start 5"); Thread.sleep(5000); println("stop 5")}
)
Future.sequence(list)
}