Akka 调度程序：等待 Runnable 完成

Question

所以我有以下 Akka 调度程序设置

implicit val testActorSystem = ActorSystem(Behaviors.empty, "test-actor-system")
implicit val executionContext = testActorSystem.executionContext

val cleaner: Runnable = () => //do something

val cancellable: Cancellable = testActorSystem.scheduler.scheduleOnce(0 seconds, cleaner)

我想等到我的Runnable 完成。我怎样才能检查这个？我知道这个任务在后台转换为Futures，但在这种情况下，我无法访问这些Futures 并等待它们。那么等待任务完成的最佳做法是什么？

Answer 1

假设您可以控制Runnable，您可以让Runnable 完成Promise 的Future：

import akka.Done
import scala.concurrent.Promise
import scala.util.control.NonFatal

val (cleaner, cleanerDoneFut) = {
  val donePromise = Promise[Done]

  val cleaner: Runnable = () => {
    try {
      // do stuff
      donePromise.success(Done)
    } catch {
      case NonFatal(e) => donePromise.failure(e)
    }
  }

  cleaner -> donePromise.future
}

val cancellable = ...

显然，这只能通过您希望只执行一次的Runnable 来完成。

请注意，您可以将任何 Runnable 包装到 Runnable 中，从而在完成时完成 Future：

// the future will contain the result of the first time the returned runnable runs
def runnableWithCompletionFuture(runnable: Runnable): (Runnable, Future[Done]) = {
  val donePromise = Promise[Done]

  val wrapped = try {
    runnable.run()
    donePromise.success(Done)
  } catch {
    case NonFatal(e) => donePromise.failure(e)
  }

  wrapped -> donePromise.future
}

def scheduleWithCompletion(when: FiniteDuration)(r: Runnable)(implicit actorSystem: ActorSystem): (Cancellable, Future[Done]) = {
  val (wrapped, fut) = runnableWithCompletionFuture(r)
  val c = actorSystem.scheduler.scheduleOnce(when, wrapped)
  c -> fut
}

通过挖掘实现，没有这样的技巧，没有办法判断任意预定的Runnable是否已经完成。

Answer 2

我认为你做不到。你拨打method：

def scheduleOnce(delay: FiniteDuration, runnable: Runnable)(implicit executor: ExecutionContext): Cancellable

而Cancellable 是：

trait Cancellable {

  /**
   * Cancels this Cancellable and returns true if that was successful.
   * If this cancellable was (concurrently) cancelled already, then this method
   * will return false although isCancelled will return true.
   *
   * Java & Scala API
   */
  def cancel(): Boolean

  /**
   * Returns true if and only if this Cancellable has been successfully cancelled
   *
   * Java & Scala API
   */
  def isCancelled: Boolean
}

所以我认为没有这样的选择。