Akka актер заблокирован другой актер работает процессор интенсивной работы

Question

У меня есть ActorSystem, который состоит из трех актеров, MasterActor, PrinterActor и BlockerActor:

PrinterActor просто спит в течение одной секунды, то печатает что-то:

public class PrinterActor extends UntypedActor { 

    @Override 
    public void onReceive(Object msg) throws Exception {   
     Thread.sleep(1000); 
     System.out.println("elapsed time is " + (System.currentTimeMillis()-(Long)msg)); 
    } 
} 

BlockerActor делает некоторые CPU-интенсивная работа:

public class BlockerActor extends UntypedActor { 

    @Override 
    public void onReceive(Object msg) throws Exception { 
     long count=0; 
     for (int i=0; i<5000; i++) { 
      for (int j=0; j<1000000000; j++) { 
       count++; 
      } 
     } 
     System.out.println("count is " + count); 
    } 
} 

MasterActor создает выше двух актеров, то говорит им, чтобы начать работать на то же самое время:

public class MasterActor extends UntypedActor { 

    @Override 
    public void onReceive(Object msg) throws Exception { 
     ActorRef printer = getContext().actorOf(Props.create(PrinterActor.class), "printerActor"); 
     ActorRef blocker = getContext().actorOf(Props.create(BlockerActor.class), "blockerActor"); 

     blocker.tell("start", getSelf()); 
     printer.tell(System.currentTimeMillis(), getSelf()); 
    } 
} 

Основной метод создает экземпляр ActorSystem, создает MasterActor под ним затем посылает актера начало сообщения.

public class App { 
    public static void main(String[] args) { 
     ActorSystem actorSystem = ActorSystem.create("Test"); 
     ActorRef master = actorSystem.actorOf(Props.create(MasterActor.class), "masterActor"); 
     master.tell("start", null); 
    } 
} 

Я ожидал, что PrinterActor завершится быстро, но это было не так. Смотрите ниже вывод:

count is 5000000000000 
elapsed time is 106856 

Она смотрит на меня PrinterActor фактически не получают отдельный поток, но делил один поток с двумя другими субъектами в системе. У меня такое впечатление, потому что если я изменять реализацию BlockerActor на:

public class BlockerActor extends UntypedActor { 

    @Override 
    public void onReceive(Object msg) throws Exception { 
     Thread.sleep(60 * 1000); 
    } 
} 

PrinterActor будет работать намного быстрее:

elapsed time is 1004 

Обратите внимание, что я не настроить любого диспетчера для моих актеров. Поэтому все они должны использовать диспетчер по умолчанию для системы, у которого есть пул из 3.0 (коэффициент параллелизма по умолчанию) * количество процессоров (8 ядер на моей машине) = 24 потока. Я даже пытался предоставить PrinterActor PinnedDispatcher (выделенный поток), но все еще не мог заставить PrinterActor ускоряться, когда работал над BlockerActor.

Теперь я действительно смущен. Разве вы не должны проявлять некоторую степень параллелизма при использовании таких актеров? Это ожидаемое поведение Акки или я сделал что-то неправильно?

PS: Я провел программу в Eclipse, с Акку 2.3.6

Answer 1

Проблема может быть с Thread.sleep, поскольку есть вопросы, связанные с платформой/реализации.

http://www.jwrapper.com/blog/worst-bug-ever-java-sleeps-3-orders-of-magnitude-too-long https://community.oracle.com/thread/2576985

Я попробовал ваш пример в Scala и не испытывал какую-либо дополнительная блокировки для PrintActor.

Вы также можете добавить другую распечатку для «внутреннего» времени выполнения, чтобы проверить, является ли проблема Thread.sleep.

истекшее время 1005 (внешний)

истекшее время 1007 (внутренний)

счетчик 5000000000

package Actors 

import akka.actor.{ActorSystem, Actor, Props} 

class PrintActor extends Actor { 
    override def receive = { 
    case externalStartTime: Long => 
     val internalStartTime = System.currentTimeMillis() 
     Thread.sleep(1000) 
     println(s"elapsed time is ${System.currentTimeMillis() - externalStartTime} (external)") 
     println(s"elapsed time is ${System.currentTimeMillis() - internalStartTime} (internal)") 
    case _ => throw new IllegalArgumentException 
    } 
} 

class BlockerActor extends Actor { 
    override def receive = { 
    case "start" => 
     var counter = 0L 
     for (i <- 1 to 5000) { 
     for (j <- 1 to 1000000) { // 1000000000 
      counter += 1 
     } 
     } 
     println(s"counter is $counter") 
    case _ => throw new IllegalArgumentException 
    } 
} 

class MasterActor extends Actor { 
    override def receive = { 
    case "start" => 
     val printer = context.actorOf(Props[PrintActor], "printerActor") 
     val blocker = context.actorOf(Props[BlockerActor], "blockerActor") 
     blocker.tell("start", self) 
     printer.tell(System.currentTimeMillis(), self) 
    case _ => throw new IllegalArgumentException 
    } 
} 

object App { 
    def main (args: Array[String]) { 
    val actorSystem = ActorSystem("Test") 
    val master = actorSystem.actorOf(Props[MasterActor], "masterActor") 
    master.tell("start", null) 
    actorSystem.shutdown() 
    } 
}