在未占用所有默认调度程序的执行元中阻止操作

Question

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

下面是我的代码：

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只是立即打印一个数字。

>　PrintActor: 44
>　PrintActor: 45

默认的调度程序是否会为阻塞操作预留一些调度程序？这样即使有这么多请求调度器的阻塞操作,系统也能处理消息。

Akka文件中的实验是否可以复制？我的配置有问题吗。

谢谢你的建议。最好的祝愿。

Answer 1

您之所以在BlockingFutuReactor中看到所有1000条打印语句之前看到PrintActor中的任何打印语句，是因为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)基本上模拟了一条至少需要五秒钟处理的消息。blockingfuturetor在处理完当前消息之前不会处理另一个消息，即使它的邮箱中有数百条消息。当BlockingFutuReactor正在处理值1的第一个Int消息时，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
...

正如您所看到的，当blockingfuturetor实际开始处理消息2时，printactor已经翻转了所有1000条消息。

如果删除第一个线程.sleep，那么您将更快地看到消息从BlockingFutuReactor的邮箱出列，因为工作被“委托”给了Future。创建future后，执行元将从其邮箱中获取下一条消息，而无需等待future完成。下面是没有第一个线程的示例输出。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
...