超时后中断 Play 2 中的长时间计算

Question

我有一个很长的计算算法，我想在 Play 应用程序中使用它。

我想添加一个超时值，所以如果计算时间超过一段时间，它应该被中断并显示一些错误消息。

查看Handling asynchronous results documentation - handling time-outs，它解释了如何在长时间计算中创建超时。

但是，我注意到虽然用户收到超时消息，但计算并没有中断，即日志消息会一直打印。

超时后如何中断长时间的计算？

示例控制器代码为：

object Application extends Controller {

  def timeout(n:Integer)  = Action.async {
    val futureInt = scala.concurrent.Future { longComputation() }
    val timeoutFuture = play.api.libs.concurrent.Promise.timeout("Oops", 1.second)
    Future.firstCompletedOf(Seq(futureInt, timeoutFuture)).map {
     case i: Int => Ok("Got result: " + i)
     case t: String => InternalServerError(t)
    }
  }

   def longComputation(): Int = {
     while (true) {
      Thread.sleep(1000)
      Logger.debug("Computing...")
    }
    return 0
   }

}

Answer 1

为了满足这个问题的要求，如果运行时间超过最大持续时间，则必须能够中断长时间运行的计算。此外，有必要处理这种中断控制器操作的可能性。

假设计算涉及多个步骤和/或重复，中断该计算的一种方法（而不是仅仅放弃其结果并让计算继续运行）是定期检查当前计算的持续时间是否大于最大持续时间.

为了明确表明此计算可能失败，可以将其声明为返回Try[T]。

然后，该操作可以在未来成功时检查计算尝试的结果，并为成功或失败的尝试生成适当的输出。

例如：

package controllers

import play.api._
import play.api.libs.concurrent.Akka
import play.api.libs.concurrent.Execution.Implicits.defaultContext
import play.api.mvc._
import play.api.Play.current
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.util._

object Application extends Controller {

  def factorial(n: Int) = Action.async {
    computeFactorial(n, 3.seconds).map { result =>
      result match {
        case Success(i) => Ok(s"$n! = $i")
        case Failure(ex) => InternalServerError(ex.getMessage)
      }
    }
  }

  def computeFactorial(n: BigInt, timeout: Duration): Future[Try[BigInt]] = {

    val startTime = System.nanoTime()
    val maxTime = timeout.toNanos

    def factorial(n: BigInt, result: BigInt = 1): BigInt = {
      // Calculate elapsed time.
      val elapsed = System.nanoTime() - startTime
      Logger.debug(s"Computing factorial($n) with $elapsed nanoseconds elapsed.")

      // Abort computation if timeout was exceeded.
      if (elapsed > maxTime) {
        Logger.debug(s"Timeout exceeded.")
        throw new ComputationTimeoutException("The maximum time for the computation was exceeded.")
      }

      // Introduce an artificial delay so that less iterations are required to produce the error.
      Thread.sleep(100)

      // Compute step.
      if (n == 0) result else factorial(n - 1, n * result)
    }

    Future {
      try {
        Success(factorial(n))
      } catch {
        case ex: Exception => Failure(ex)
      }
    }(Contexts.computationContext)
  }

}

class ComputationTimeoutException(msg: String) extends RuntimeException(msg)

object Contexts {
  implicit val computationContext: ExecutionContext = Akka.system.dispatchers.lookup("contexts.computationContext")
}

如果不需要将计算结果显式标记为错误，并且 Play 的默认异步故障处理（返回 500 Internal Server Error）就足够了，则代码可以更简洁：

object Application extends Controller {

  def factorial(n: Int) = Action.async {
    computeFactorial(n, 3.seconds).map { i => Ok(s"$n! = $i") }
  }

  def computeFactorial(n: BigInt, timeout: Duration): Future[BigInt] = {
    val startTime = System.nanoTime()
    val maxTime = timeout.toNanos

    def factorial(n: BigInt, result: BigInt = 1): BigInt = {
      if (System.nanoTime() - startTime > maxTime) {
        throw new RuntimeException("The maximum time for the computation was exceeded.")
      }
      Thread.sleep(100)
      if (n == 0) result else factorial(n - 1, n * result)
    }

    Future { factorial(n) }(Akka.system.dispatchers.lookup("contexts.computationContext"))
  }

}

示例在自定义上下文中运行计算，该上下文提供了一个不同于 Play 用于处理 HTTP 请求的线程池的线程池。请参阅Understanding Play thread pools 了解更多信息。上下文在application.conf中声明：

contexts {
  computationContext {
    fork-join-executor {
      parallelism-factor=20
      parallelism-max = 200
    }
  }
}

有关可下载的示例，请参阅 this GitHub project。