如何通过超时取消链式 Scala 未来?

Question

假设我有一个定义的未来，如下所示：

import scala.concurrent.Future
def appendCharWithTimeout(transformationId: String, char: Char, delayTimeMs: Long, delayTimes: Int) = (s: String) => {
for (i <- 1 to delayTimes) {
println(s"$transformationId waiting iteration $i ...")
Thread.sleep(delayTimeMs)
}
s"$s$char"
}
Future("Hello ")
.map( appendCharWithTimeout("mapJ", 'J', 200, 5) )
.map( appendCharWithTimeout("mapO", 'o', 200, 5) )
.map( appendCharWithTimeout("mapH", 'h', 200, 5) )
.map( appendCharWithTimeout("mapN", 'n', 200, 5) )
.map( appendCharWithTimeout("map!", '!', 200, 5) )

这个未来的执行时间为5秒(5 * 5 * 200ms(。

我正在寻找一种方法来将这个未来包装在某种"超时上下文"中，并通过超时停止执行，因此并非所有转换都会被执行。

理想情况下，我设想有这样的东西：

Future("Hello ")
.within(2 seconds)
.map( appendCharWithTimeout("mapJ", 'J', 200, 5) )
.map( appendCharWithTimeout("mapO", 'o', 200, 5) )
.map( appendCharWithTimeout("mapH", 'h', 200, 5) )
.map( appendCharWithTimeout("mapN", 'n', 200, 5) )
.map( appendCharWithTimeout("map!", '!', 200, 5) )

输出应为：

mapJ waiting iteration 1 ...
mapJ waiting iteration 2 ...
mapJ waiting iteration 3 ...
mapJ waiting iteration 4 ...
mapJ waiting iteration 5 ...
mapO waiting iteration 1 ...
mapO waiting iteration 2 ...
mapO waiting iteration 3 ...
mapO waiting iteration 4 ...
mapO waiting iteration 5 ...

Answer 1

以下是执行此操作的几种方法：

0(不要链接Futures。执行是顺序的，因此只需在单个Future中使用循环并跟踪循环中的总运行时间。

1( 在Future之外的val中记录开始时间，并使用它来修改给定appendCharWithTimeout的超时值，以便不超过总执行时间。

2(appendCharWithTimeout总执行时间，并将剩余时间返回到下一次迭代。使用此选项可在超过超时时停止执行。

选择取决于实际代码的实际作用，以及是否可以在appendCharWithTimeout中更改代码。

Answer 2

首先，请不要将Thread.sleep与期货混为一谈。期货与 ExecutionContext 一起工作，该上下文在线程池上调度计算。所以，如果你的未来要阻止上述线程......这将导致问题。

import java.util.{Timer, TimerTask}
import scala.concurrent.{Future, Promise}
import scala.concurrent.duration.{Duration, TimeUnit}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success}

def createFutureWithDelay[T](result: T, timeout: Duration) = {
val promise = Promise[T]
val timerTask = new TimerTask {
override def run(): Unit = promise.success(result)
}
val timer = new Timer()
timer.schedule(timerTask, timeout.toMillis)
promise.future
}
def getNonBlockingFutureWithTimeout[A, T](computation: => T, timeout: Duration, t: Throwable) = {
val promise = Promise[T]
promise.tryCompleteWith(Future(computation))
val timerTask = new TimerTask {
override def run(): Unit = promise.failure(t)
}
val timer = new Timer()
timer.schedule(timerTask, timeout.toMillis)
promise.future
}
def wrapFutureWithTimeout[T](f: Future[T], timeout: Duration, t: Throwable) = {
val promise = Promise[T]
promise.tryCompleteWith(f)
val timerTask = new TimerTask {
override def run(): Unit = promise.failure(t)
}
val timer = new Timer()
timer.schedule(timerTask, timeout.toMillis)
promise.future
}
val f = createFutureWithDelay(5, 5 minutes).flatMap(_ => createFutureWithDelay(5, 5 minutes))
val f2 = wrapFutureWithTimeout(f, 5 seconds, new Throwable("ENDING with timeout"))
f2.onComplete({
case Success(value) => println(s"success - $value")
case Failure(t) => println(s"failure - ${t.getMessage}")
})