Scala之Future超时

Posted 2021-01-08 junjiang3

最近在开发中使用akka http进行请求，返回的是一个future，并且要对future进行超时设置，不知怎么设置，因此学习了下。

一、Future阻塞

首先，scala中的future不支持内置超时，要想达到这样的目的，可以使用Await进行阻塞，具体例子如下：

import scala.concurrent._
import scala.concurrent.duration._
import ExecutionContext.Implicits.global

lazy val f = future { Thread.sleep(2000); true }
Await.result(f, 1 second)

上面的代码将超时，报如下错误：

java.util.concurrent.TimeoutException：
    at scala.concurrent.impl.Promise $ DefaultPromise.ready（Promise.scala：219）
    at scala.concurrent.impl.Promise $ DefaultPromise.result（Promise.scala：223）
    at scala.concurrent.Await $$ anonfun $ result $ 1.apply（package.scala：107）
    at scala.concurrent.BlockContext $ DefaultBlockContext $ .blockOn（BlockContext.scala：53） 
...

二、非阻塞Future超时

但是，我们知道，在future上设置阻塞不是官网推荐的一种方式，因为这会浪费一个线程。因此，我们可以使用akka after实现一种非阻塞式的future超时:

import scala.concurrent._
import scala.concurrent.duration._
import ExecutionContext.Implicits.global
import scala.util.{Failure, Success}
import akka.actor.ActorSystem
import akka.pattern.after

val system = ActorSystem("theSystem")

lazy val f = future { Thread.sleep(2000); true }
  lazy val t = after(duration = 1 second, using = system.scheduler)(Future.failed(new TimeoutException("Future timed out!")))

  val fWithTimeout = Future firstCompletedOf Seq(f, t)

  fWithTimeout.onComplete {
  case Success(x) => println(x)
  case Failure(error) => println(error)
}

但是，注意了，为了确保在执行前，计时还没有开始，必须将after设置lazy val。

但是上述这种模式的缺点是它依赖于akka，因此，我们可以使用纯scala模式，来模仿实现after的功能

为了更容易使用future的超时设置，我们可以使用隐式类来扩展scala future从而支持超时：

import scala.concurrent._
import scala.concurrent.duration.FiniteDuration
import ExecutionContext.Implicits.global
import akka.actor.ActorSystem
import akka.pattern.after

implicit class FutureExtensions[T](f: Future[T]) {
  def withTimeout(timeout: => Throwable)(implicit duration: FiniteDuration, system: ActorSystem): Future[T] = {
    Future firstCompletedOf Seq(f, after(duration, system.scheduler)(Future.failed(timeout)))
  }
}

现在，我们可以随时很方便的给future设置超时了：

import scala.concurrent._
import scala.concurrent.duration._
import scala.util.{ Success, Failure }
import ExecutionContext.Implicits.global
import akka.actor.ActorSystem

implicit val system = ActorSystem("theSystem")
implicit val timeout = 1 second

lazy val f = future { Thread.sleep(2000); true }

  f withTimeout new TimeoutException("Future timed out!") onComplete {
  case Success(x) => println(x)
  case Failure(error) => println(error)
}