scala - 由于某种原因无法从 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。)