不占用所有默认调度程序的 Actor 中的阻塞操作

Question

我最近在学习 Akka Actor。我在 Actor 中阅读了调度员的文档。我很好奇演员中的阻塞操作。文档中的最后一个topic 描述了如何解决问题。我正在尝试重现文档中的示例实验。

这是我的代码：

package dispatcher

import akka.actor.{ActorSystem, Props}
import com.typesafe.config.ConfigFactory

object Main extends App{

  var config = ConfigFactory.parseString(
    """
      |my-dispatcher{
      |type = Dispatcher
      |
      |executor = "fork-join-executor"
      |
      |fork-join-executor{
      |fixed-pool-size = 32
      |}
      |throughput = 1
      |}
    """.stripMargin)

//  val system = ActorSystem("block", ConfigFactory.load("/Users/jiexray/IdeaProjects/ActorDemo/application.conf"))


  val system = ActorSystem("block")


  val actor1 = system.actorOf(Props(new BlockingFutureActor()))
  val actor2 = system.actorOf(Props(new PrintActor()))

  for(i <- 1 to 1000){
    actor1 ! i
    actor2 ! i
  }

}

package dispatcher

import akka.actor.Actor

import scala.concurrent.{ExecutionContext, Future}

class BlockingFutureActor extends Actor{
  override def receive: Receive = {
    case i: Int =>
      Thread.sleep(5000)
      implicit val excutionContext: ExecutionContext = context.dispatcher
      Future {
        Thread.sleep(5000)
        println(s"Blocking future finished ${i}")
      }
  }
}
package dispatcher

import akka.actor.Actor

class PrintActor extends Actor{
  override def receive: Receive = {
    case i: Int =>
      println(s"PrintActor: ${i}")
  }
}

我只是用默认调度程序创建了一个ActorSystem，所有参与者都依赖于这些。 BlockingFutureActor 有一个封装在Future 中的阻塞操作。 PrintActor 只是立即打印一个数字。

在文档的解释中，默认的dispatcher会被BlockingFutureActor中的Futures占用，从而导致PrintActor的消息阻塞。应用程序卡在某个地方，例如：

>　PrintActor: 44
>　PrintActor: 45

很遗憾，我的代码没有被阻止。 PrintActor 的所有输出都顺利显示。但是BlockingFutureActor 的输出就像挤牙膏一样。我尝试通过 Intellij 的 Debug 监控我的线程信息，我得到：

您可能会发现只有两个调度员在睡觉（BlockingFutureActor 使这种情况发生）。其他人正在等待，这意味着他们可以进行新消息传递。

我已经阅读了关于 Actor(page) 中阻塞操作的答案。引用说“调度程序实际上是线程池。将两者分开可以保证缓慢的阻塞操作不会使另一个饿死。这种方法通常被称为批量标题，因为这个想法是如果应用程序的一部分发生故障，其余部分仍保持响应。”

默认调度程序是否会为阻塞操作留出一些调度程序？这样即使有这么多阻塞操作要求调度程序，系统也可以处理消息。

Akka文档中的实验可以复现吗？是不是我的配置有问题。

感谢您的建议。最好的祝福。

Answer 1

您在来自BlockingFutureActor 的任何打印语句之前看到来自PrintActor 的所有1000 个打印语句的原因是因为BlockingFutureActor 的receive 块中的第一个Thread.sleep 调用。这个Thread.sleep 是您的代码与官方文档中的示例之间的关键区别：

override def receive: Receive = {
  case i: Int =>
    Thread.sleep(5000) // <----- this call is not in the example in the official docs
    implicit val excutionContext: ExecutionContext = context.dispatcher
    Future {
      ...
    }
}

请记住，参与者一次处理一条消息。 Thread.sleep(5000) 基本上模拟了一条至少需要五秒钟才能处理的消息。 BlockingFutureActor 在处理完当前消息之前不会处理另一条消息，即使它的邮箱中有数百条消息。当BlockingFutureActor 正在处理第一个Int 值1 的消息时，PrintActor 已经完成了对发送给它的所有 1000 条消息的处理。为了更清楚地说明这一点，让我们添加一个println 声明：

override def receive: Receive = {
  case i: Int =>
    println(s"Entering BlockingFutureActor's receive: $i") // <-----
    Thread.sleep(5000)
    implicit val excutionContext: ExecutionContext = context.dispatcher
    Future {
      ...
    }
}

我们运行程序时的示例输出：

Entering BlockingFutureActor's receive: 1
PrintActor: 1
PrintActor: 2
PrintActor: 3
...
PrintActor: 1000
Entering BlockingFutureActor's receive: 2
Entering BlockingFutureActor's receive: 3
Blocking future finished 1
...

如您所见，当BlockingFutureActor 真正开始处理消息2 时，PrintActor 已经处理了所有 1000 条消息。

如果您删除第一个Thread.sleep，那么您会更快地看到邮件从BlockingFutureActor 的邮箱中出列，因为该工作正在“委托”给Future。一旦创建了Future，actor 就会从其邮箱中获取下一条消息，而无需等待Future 完成。下面是没有第一个 Thread.sleep 的示例输出（每次运行它都不会完全相同）：

Entering BlockingFutureActor's receive: 1
PrintActor: 1
PrintActor: 2
...
PrintActor: 84
PrintActor: 85
Entering BlockingFutureActor's receive: 2
Entering BlockingFutureActor's receive: 3
Entering BlockingFutureActor's receive: 4
Entering BlockingFutureActor's receive: 5
PrintActor: 86
PrintActor: 87
...