下面的代码中有一个错误（尝试使用 Akka 递归）。算法停止并且进程（Java 应用程序）在 JVM 中永远执行，除非我从系统监视器中杀死它。我相信修复它应该是一个非常简单的技巧。
这是一个关于如何使用 Akka 进行并行 Pi 近似的示例。下面是展示 Akka 如何与递归 Actor 一起工作的尝试。所以主人创建了 2 个工人，向他们发送相同的消息以减少一些int值。他们并行执行此操作并检查整数值是否不等于 0。如果是，则将结果整数值 (0) 返回给 master，或者他们都再次创建 2 个 worker 并向它们发送最近递减的值。如果这棵树的深度大于 1（整数大于 1 值），然后工作人员将他们的结果发送给调用他们的工作人员，最后只发送给主人。嗯，这真的很简单（Decrement、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);
}
}
}
}