由于某种原因无法从 RejectedExecutionException 中恢复

Question

所以，我正在尝试编写一个一直运行直到你告诉它停止的任务：

import scala.concurrent.Future
import scala.concurrent.ExecutionContext
import java.util.concurrent.RejectedExecutionException

def runUntilShutdown(f: => Unit) = {
  val ctx = ExecutionContext.fromExecutorService(null)
  import ExecutionContext.global
  def runTask(): Future[Unit] = Future(f)(ctx)
    .flatMap(_ => runTask())(ctx)
  runTask()
    .recover { case _: RejectedExecutionException => () }(global)
    .onComplete { _ => println("Done") }(global)
  ctx
}

val ctx = runUntilShutdown(Thread.sleep(1000))
ctx.shutdown

我希望这只是在最后打印“完成”，但这永远不会发生。

相反，RejectedExecutionException 的堆栈跟踪被转储到 stderr：

java.util.concurrent.RejectedExecutionException
at scala.concurrent.forkjoin.ForkJoinPool.fullExternalPush(ForkJoinPool.java:1870)
at scala.concurrent.forkjoin.ForkJoinPool.externalPush(ForkJoinPool.java:1834)
at scala.concurrent.forkjoin.ForkJoinPool.execute(ForkJoinPool.java:2973)
at scala.concurrent.impl.ExecutionContextImpl$$anon$1.execute(ExecutionContextImpl.scala:136)
at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44)
at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252)
at scala.concurrent.Promise$class.complete(Promise.scala:55)
at scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:157)
at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:23)
at scala.concurrent.forkjoin.ForkJoinTask$AdaptedRunnableAction.exec(ForkJoinTask.java:1361)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

想法？

Answer 1

此问题已在 Scala 2.13* 中通过 Future & Promise 的新实现得到修复， 您可以在 Scala 2.13.0-M5 上尝试您的示例，但您必须在您的 EC 上调用 shutdownNow 否则它将继续运行，因为它不会接受新任务，但它已经在运行您的 Future .

示例输出：

Welcome to Scala 2.13.0-20181205-121558-76b34c4 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_144).
Type in expressions for evaluation. Or try :help.

scala> import scala.concurrent.Future
import scala.concurrent.Future

scala> import scala.concurrent.ExecutionContext
import scala.concurrent.ExecutionContext

scala> import java.util.concurrent.RejectedExecutionException
import java.util.concurrent.RejectedExecutionException

scala> def runUntilShutdown(f: => Unit) = {
     |   val ctx = ExecutionContext.fromExecutorService(null)
     |   import ExecutionContext.global
     |   def runTask(): Future[Unit] = Future(f)(ctx)
     |     .flatMap(_ => runTask())(ctx)
     |   runTask()
     |     .recover { case _: RejectedExecutionException => () }(global)
     |     .onComplete { _ => println("Done") }(global)
     |   ctx
     | }
runUntilShutdown: (f: => Unit)scala.concurrent.ExecutionContextExecutorService

scala> val ctx = runUntilShutdown(Thread.sleep(1000))
ctx: scala.concurrent.ExecutionContextExecutorService = scala.concurrent.impl.ExecutionContextImpl$$anon$3@23d060c2[Running, parallelism = 8, size = 1, active = 1, running = 0, steals = 0, tasks = 0, submissions = 0]

scala> ctx.shutdownNow
res2: java.util.List[Runnable] = []

scala> Done

*: https://github.com/scala/bug/issues/9071（无法通过 Future & Promise 的旧实现来实现正确的行为，因此目前没有计划向后移植到 2.12。）

Answer 2

看起来你有一个永远不会结束的递归def runTask(): Future[Unit] = Future(f)(ctx).flatMap(_ => runTask())(ctx)。因此，不会打印Done 的输出，因为未来永远不会完成。

本例正确处理异常

  import scala.concurrent.Future
  import scala.concurrent.ExecutionContext
  import java.util.concurrent.RejectedExecutionException
  import java.util.concurrent.Executors

  def runUntilShutdown(f: () => Unit) = {
    implicit val ctx = ExecutionContext.fromExecutorService(Executors.newCachedThreadPool())
    Future { f() }
      .recover { case _: RejectedExecutionException => () }
      .onComplete { _ =>
        println("Done")
      }
    ctx
  }

  val ctx = runUntilShutdown { () =>
    Thread.sleep(10000)
  }
  ctx.shutdown()