Play框架期货未由默认调度程序并行化

Question

[我正在尝试测试Play应用中的ExecutionContext行为，发现使用默认分派器通过调用as.dispatcher，as.dispatchers.lookup("akka.actor.default-dispatcher")无法实现任何程度的并行度或将默认执行上下文作为参数传递给我的Controller类：

class HomeController @Inject()(cc: ControllerComponents)(implicit ec: ExecutionContext)

我以here中提供的播放示例为基础。并添加/更改以下配置：

路线

GET    /futures    controllers.HomeController.testFutures(dispatcherId: String)

common.conf

akka {
  my-dispatcher {
    executor = "fork-join-executor"
    fork-join-executor {
      # vm-cores = 4
      parallelism-min = 4

      parallelism-factor = 2.0

      # 2x vm-cores
      parallelism-max = 8
    }
  }

  actor.default-dispatcher {
    executor = "fork-join-executor"
    fork-join-executor {
      # vm-cores = 4
      parallelism-min = 4

      parallelism-factor = 2.0

      # 2x vm-cores
      parallelism-max = 8
    }
  }
}

HomeController

@Singleton
class HomeController @Inject()(cc: ControllerComponents, as: ActorSystem) extends AbstractController(cc) {
  import HomeController._

  def testFutures(dispatcherId: String) = Action.async { implicit request =>
    implicit val dispatcher = as.dispatchers.lookup(dispatcherId)
    Future.sequence((0 to 10).map(i => Future {
      val time = 1000 + Random.nextInt(200)
      log.info(s"Sleeping #$i for $time ms")
      Thread.sleep(time)
      log.info(s"Awakening #$i")
    })).map(_ => Ok("ok"))
  }
}

由于某种原因，对http://localhost:9000/futures?dispatcherId=akka.actor.default-dispatcher（默认调度程序的调用不会并行化并产生以下输出：

[info] c.HomeController - Sleeping #0 for 1044 ms
[info] c.HomeController - Awakening #0
[info] c.HomeController - Sleeping #1 for 1034 ms
[info] c.HomeController - Awakening #1
[info] c.HomeController - Sleeping #2 for 1031 ms
[info] c.HomeController - Awakening #2
[info] c.HomeController - Sleeping #3 for 1065 ms
[info] c.HomeController - Awakening #3
[info] c.HomeController - Sleeping #4 for 1082 ms
[info] c.HomeController - Awakening #4
[info] c.HomeController - Sleeping #5 for 1057 ms
[info] c.HomeController - Awakening #5
[info] c.HomeController - Sleeping #6 for 1090 ms
[info] c.HomeController - Awakening #6
[info] c.HomeController - Sleeping #7 for 1165 ms
[info] c.HomeController - Awakening #7
[info] c.HomeController - Sleeping #8 for 1173 ms
[info] c.HomeController - Awakening #8
[info] c.HomeController - Sleeping #9 for 1034 ms
[info] c.HomeController - Awakening #9
[info] c.HomeController - Sleeping #10 for 1056 ms
[info] c.HomeController - Awakening #10

但是调用此http://localhost:9000/futures?dispatcherId=akka.my-dispatcher（使用另一个调度程序）并行化正确性并产生以下输出。

[info] c.HomeController - Sleeping #1 for 1191 ms
[info] c.HomeController - Sleeping #0 for 1055 ms
[info] c.HomeController - Sleeping #7 for 1196 ms
[info] c.HomeController - Sleeping #4 for 1121 ms
[info] c.HomeController - Sleeping #6 for 1040 ms
[info] c.HomeController - Sleeping #2 for 1016 ms
[info] c.HomeController - Sleeping #5 for 1107 ms
[info] c.HomeController - Sleeping #3 for 1165 ms
[info] c.HomeController - Awakening #2
[info] c.HomeController - Sleeping #8 for 1002 ms
[info] c.HomeController - Awakening #6
[info] c.HomeController - Sleeping #9 for 1127 ms
[info] c.HomeController - Awakening #0
[info] c.HomeController - Sleeping #10 for 1016 ms
[info] c.HomeController - Awakening #5
[info] c.HomeController - Awakening #4
[info] c.HomeController - Awakening #3
[info] c.HomeController - Awakening #1
[info] c.HomeController - Awakening #7
[info] c.HomeController - Awakening #8
[info] c.HomeController - Awakening #10
[info] c.HomeController - Awakening #9

任何想法为什么会发生这种情况？

Answer 1

[我认为该行为由类型为akka.actor.default-dispatcher的BatchingExecutor给出，并且将在诸如map/flatmap之类的操作中尝试通过在同一线程中执行它们来避免不必要的调度，从而进行优化。在我们要阻塞的情况下，我们可以用scala.concurrent.blocking (Thread.sleep (time))作为提示来表示它，这样，在ThreadLocal[BlockContext]中存储了一个标记，该标记指示要阻塞的意图，并且不应用优化而是将操作抛出另一个线程。

如果将此行Thread.sleep(time)更改为此scala.concurrent.blocking(Thread.sleep(time))，则将获得所需的行为

@Singleton
class HomeController @Inject()(cc: ControllerComponents, as: ActorSystem) extends AbstractController(cc) {
  import HomeController._

  def testFutures(dispatcherId: String) = Action.async { implicit request =>
    implicit val dispatcher = as.dispatchers.lookup(dispatcherId)
    Future.sequence((0 to 10).map(i => Future {
      val time = 1000 + Random.nextInt(200)
      log.info(s"Sleeping #$i for $time ms")
      scala.concurrent.blocking(Thread.sleep(time))
      log.info(s"Awakening #$i")
    })).map(_ => Ok("ok"))
  }
}

[info] play.api.Play - Application started (Dev) (no global state)
Sleeping #0 for 1062 ms
Sleeping #1 for 1128 ms
Sleeping #2 for 1189 ms
Sleeping #3 for 1105 ms
Sleeping #4 for 1169 ms
Sleeping #5 for 1178 ms
Sleeping #6 for 1057 ms
Sleeping #7 for 1003 ms
Sleeping #8 for 1164 ms
Sleeping #9 for 1029 ms
Sleeping #10 for 1005 ms
Awakening #7
Awakening #10
Awakening #9
Awakening #6
Awakening #0
Awakening #3
Awakening #1
Awakening #8
Awakening #4
Awakening #5
Awakening #2