使用Akka遞歸遞減int值（示例）

Question

下面的代碼中某處存在一個錯誤（嘗試使用Akka遞歸）。算法停止，並且進程（Java應用程序）在JVM中永遠執行，除非我從系統監視器中殺死它。我相信這應該是一個非常簡單的解決辦法。

Here is an example關於如何使用Akka進行平行Pi逼近。以下是試圖展示Akka如何與遞歸演員合作。因此，主人創建2個工人，向他們發送相同的消息以減少一些int的值。他們並行地執行該操作，並檢查整數值是否不等於0.如果是，則返回結果整數值（0）給主人，或者他們都再次創建2個工作人員並向他們發送最近遞減的值。如果這棵樹的深度大於1（整數大於1的值），那麼工作人員將他們的結果發送給調用它們的工作人員，並且只在主人結束時。嗯，這是很容易爲以下（遞減，NewIntValue和FinalIntValue基本上是相同的，他們有不同的名稱，以使其更容易理解）：

import akka.actor.ActorRef; 
import akka.actor.ActorSystem; 
import akka.actor.Props; 
import akka.actor.UntypedActor; 
import akka.actor.UntypedActorFactory; 
import akka.routing.RoundRobinRouter; 

public class StackOverFlow { 

    public static void main(String[] args) { 
     StackOverFlow rid = new StackOverFlow(); 
     rid.start(2); 
    } 

    public void start(final int workersNumber) { 
     // create an Akka system 
     ActorSystem system = ActorSystem.create("IntDec"); 
     // create the result listener, which will print the result and shutdown the system 
     final ActorRef listener = system.actorOf(new Props(Listener.class), "listener"); 
     // create the master 
     ActorRef master = system.actorOf(new Props(new UntypedActorFactory() { 
      public UntypedActor create() { 
       return new Master(workersNumber, listener); 
      } 
     }), "master"); 
     // start the computation 
     master.tell(new Compute()); 
    } 

    static class Compute {} 

    static class Decrement { 
     private final int intValue; 
     public Decrement(int value) { 
      this.intValue = value; 
     } 
     public int getValue() { 
      return intValue; 
     } 
    } 

    static class NewIntValue { 
     private final int intValue; 
     public NewIntValue(int value) { 
      intValue = value; 
     } 
     public int getValue() { 
      return intValue; 
     } 
    } 

    static class FinalIntValue { 
     private final int intValue; 
     public FinalIntValue(int value) { 
      intValue = value; 
     } 
     public int getValue() { 
      return intValue; 
     } 
    } 

    public static class Worker extends UntypedActor { 

     private int resultsNumber = 0; 
     private final int messagesNumber = 2; 

     private final ActorRef workerRouter; 

     public Worker(final int workersNumber) { 

      workerRouter = getContext().actorOf(
        new Props(new UntypedActorFactory() { 
         public UntypedActor create() { 
          return new Worker(workersNumber); 
         } 
        }).withRouter(
         new RoundRobinRouter(workersNumber) 
        ), "workerRouter"); 

     } 

     public void onReceive(Object message) { 

      if (message instanceof Decrement) { 
       // get and decrement the int value 
       Decrement job = (Decrement) message; 
       int intValue = job.getValue(); 
       System.out.println("\tWorker:Decrement " + intValue); 
       intValue--; 
       if (intValue == 0) { 
        // we are finished 
        getSender().tell(new NewIntValue(intValue), getSelf()); 
        // stop this actor and all its supervised children 
        getContext().stop(getSelf()); 
       } else { 
        for (int i = 0; i < messagesNumber; i++) { 
         // notify a worker 
         workerRouter.tell(new Decrement(intValue), getSelf()); 
        } 
       } 

      } else if (message instanceof NewIntValue) { 

       NewIntValue newInt = (NewIntValue) message; 
       int intValue = newInt.getValue(); 

       System.out.println("\tWorker:NewIntValue!!! " + intValue); 

       resultsNumber++; 
       if (resultsNumber == messagesNumber) { 
        // we are finished 
        getSender().tell(new NewIntValue(intValue), getSelf()); 
        // stop this actor and all its supervised children 
        getContext().stop(getSelf()); 
       } 

      } else unhandled(message); 
     } 

    } 

    public static class Master extends UntypedActor { 

     private int resultsNumber = 0; 
     private final int messagesNumber = 2; 

     private int intValue = 2; 

     private final ActorRef listener; 
     private final ActorRef workerRouter; 

     public Master(final int workersNumber, ActorRef listener) { 

      this.listener = listener; 

      workerRouter = getContext().actorOf(
        new Props(new UntypedActorFactory() { 
         public UntypedActor create() { 
          return new Worker(workersNumber); 
         } 
        }).withRouter(
         new RoundRobinRouter(workersNumber) 
        ), "workerRouter"); 

     } 

     public void onReceive(Object message) { 

      if (message instanceof Compute) { 

       System.out.println("\tMaster:Compute " + intValue); 

       System.out.println(
         "\n\tInitial integer value: " + intValue); 

       for (int i = 0; i < messagesNumber; i++) { 
        workerRouter.tell(new Decrement(intValue), getSelf()); 
       } 

      } else if (message instanceof NewIntValue) { 

       NewIntValue newInt = (NewIntValue) message; 
       intValue = newInt.getValue(); 

       System.out.println("\tMaster:NewIntValue " + intValue); 

       resultsNumber++; 
       if (resultsNumber == messagesNumber) { 
        // send the result to the listener 
        listener.tell(new FinalIntValue(intValue), getSelf()); 
        // stop this actor and all its supervised children 
        getContext().stop(getSelf()); 
       } 

      } else unhandled(message); 

     } 

    } 

    public static class Listener extends UntypedActor { 

     public void onReceive(Object message) { 

      if (message instanceof FinalIntValue) { 
       FinalIntValue finalInt = (FinalIntValue) message; 
       System.out.println(
         "\n\tFinal integer value: " + finalInt.getValue()); 
       getContext().system().shutdown(); 
      } else { 
       unhandled(message); 
      } 

     } 

    } 

} 

Answer 1

1. 添加private ActorRef sender;到Worker類;
2. add sender = getSender();在Decrement消息的開頭;
3. 更改getSender()到sender在NewIntValue方法的 Worker類;