11

这是我之前的问题的后续。

假设我有一个任务，它执行一个可中断的阻塞调用。我想将它作为a运行Future并使用.failurePromise

我希望取消工作如下：

  • 如果在任务完成之前取消任务，我希望任务“立即”完成，如果它已经开始，则中断阻塞调用，我希望调用.FutureonFailure

  • 如果在任务完成后取消任务，我想得到一个状态，说明取消失败，因为任务已经完成。

是否有意义？是否可以在 Scala 中实现？有没有这种实现的例子？

4

4 回答 4

13

scala.concurrent.Future 是只读的，因此一个读者不能为其他读者搞砸事情。

看起来你应该能够实现你想要的如下：

def cancellableFuture[T](fun: Future[T] => T)(implicit ex: ExecutionContext): (Future[T], () => Boolean) = {
  val p = Promise[T]()
  val f = p.future
  p tryCompleteWith Future(fun(f))
  (f, () => p.tryFailure(new CancellationException))
}

val (f, cancel) = cancellableFuture( future => {
  while(!future.isCompleted) continueCalculation // isCompleted acts as our interrupted-flag

  result  // when we're done, return some result
})

val wasCancelled = cancel() // cancels the Future (sets its result to be a CancellationException conditionally)
于 2013-04-16T00:43:12.523 回答
11

这是 Victor 的代码的可中断版本，根据他的评论（Victor，如果我误解了，请纠正我）。

object CancellableFuture extends App {

  def interruptableFuture[T](fun: () => T)(implicit ex: ExecutionContext): (Future[T], () => Boolean) = {
    val p = Promise[T]()
    val f = p.future
    val aref = new AtomicReference[Thread](null)
    p tryCompleteWith Future {
      val thread = Thread.currentThread
      aref.synchronized { aref.set(thread) }
      try fun() finally {
        val wasInterrupted = (aref.synchronized { aref getAndSet null }) ne thread
        //Deal with interrupted flag of this thread in desired
      }
    }

    (f, () => {
      aref.synchronized { Option(aref getAndSet null) foreach { _.interrupt() } }
      p.tryFailure(new CancellationException)
    })
  }

  val (f, cancel) = interruptableFuture[Int] { () =>
    val latch = new CountDownLatch(1)

    latch.await(5, TimeUnit.SECONDS)    // Blocks for 5 sec, is interruptable
    println("latch timed out")

    42  // Completed
  }

  f.onFailure { case ex => println(ex.getClass) }
  f.onSuccess { case i => println(i) }

  Thread.sleep(6000)   // Set to less than 5000 to cancel

  val wasCancelled = cancel()

  println("wasCancelled: " + wasCancelled)
}

输出Thread.sleep(6000)是：

latch timed out
42
wasCancelled: false

输出Thread.sleep(1000)是：

wasCancelled: true
class java.util.concurrent.CancellationException
于 2013-04-17T02:49:30.670 回答
6

Twitter 的期货实施取消。看看这里：

https://github.com/twitter/util/blob/master/util-core/src/main/scala/com/twitter/util/Future.scala

第 563 行显示了负责此操作的抽象方法。Scala 的期货目前不支持取消。

于 2013-04-16T01:20:51.487 回答
2

您可以使用 Monix 库而不是 Future

https://monix.io

于 2018-05-26T01:19:10.353 回答