在 Scala 中执行后台操作的方法

Question

我想在后台做一些耗时的任务。所以我需要在不同的线程中开始计算，能够检查它是否完成（可能失败）并且能够在不需要时中止计算。计算结束后应该调用同步回调函数来存储计算值。

它可以被编程为 Thread 类的一些包装器。但我想这个基本功能已经在一些 scala 库中实现了。我试图搜索但只找到对我的简单任务来说太多的 Akka。 scala.concurrent.ExecutionContext 有有用的 execute 方法，但它不返回任何对象来检查计算状态并按需中止它。

什么库包含已经描述的功能？

---

我检查了scala.concurrent.Future。它缺乏中止计算的能力，这是至关重要的。我使用以下策略：在后台计算一些消耗函数并提供合理的默认值。如果函数的参数发生变化，我会放弃原来的计算并开始新的计算。我无法想象如何用 Future.flatMap 重写这个策略。

Answer 1

既然您要求取消，我将演示如何通过 Twitter's implementation 使用期货：

import com.twitter.util.{ Await, Future, FuturePool }

def computeFast(i: Int) = { Thread.sleep(1000); i + 1 }
def computeSlow(i: Int) = { Thread.sleep(1000000); i + 1 }

val fastComputation = FuturePool.unboundedPool(computeFast(1))
val slowComputation = FuturePool.unboundedPool(computeSlow(1))

现在您可以轮询结果：

scala> fastComputation.poll
res0: Option[com.twitter.util.Try[Int]] = Some(Return(2))

scala> slowComputation.poll
res1: Option[com.twitter.util.Try[Int]] = None

或者设置回调：

fastComputation.onSuccess(println)
slowComputation.onFailure(println)

不过，大多数时候最好使用map 和flatMap 来描述如何组合计算。

取消有点复杂（这只是一个演示——您需要提供自己的取消逻辑）：

import com.twitter.util.Promise

def cancellableComputation(i: Int): Future[Int] = {
  val p = Promise[Int]

  p.setInterruptHandler {
    case t =>
      println("Cancelling the computation")
      p.setException(t)
  }

  FuturePool.unboundedPool(computeSlow(i)).onSuccess(p.setValue)
  p
}

然后：

scala> val myFuture = cancellableComputation(10)
myFuture: com.twitter.util.Future[Int] = Promise@129588027(state=Interruptible(List(),<function1>))

scala> myFuture.poll
res4: Option[com.twitter.util.Try[Int]] = None

scala> myFuture.raise(new Exception("Stop this thing"))
Cancelling the computation

scala> myFuture.poll
res6: Option[com.twitter.util.Try[Int]] = Some(Throw(java.lang.Exception: Stop this thing))

您可能可以对标准库的未来做类似的事情。