java - Pi 近似中的死字母 (java)

Question

我正在尝试使用 Akka (2.5.8) 计算 Pi。我已经更改了旧版本的代码(将 UntypedActor 更改为 AbstractActor，使用了 ReceiveBuilder 等)，但输出中有问题 - pi 值乘以 2。更何况，那里还有死信。在旧版本的代码中，Listener 类中有一个 getContext().system().shutdown() 命令，我无法在这个版本中替换它(是有可能通过使用不同的东西吗？)。这可能是它们发生的原因，但我想确保。

我尝试了什么？我发现有一个CooperativeShutdown.apply( ActorSystem )方法，并在告诉master计算后尝试使用它。我没能用这种方式解决这个问题。

package org.example;

import akka.actor.*;
import akka.japi.pf.ReceiveBuilder;
import akka.routing.ActorRefRoutee;
import akka.routing.RoundRobinRoutingLogic;
import akka.routing.Routee;
import akka.routing.Router;

import java.util.ArrayList;
import java.util.List;

public class Pi {

  public static void main(String[] args) {

    Pi pi = new Pi();

    int nrOfWorkers = 2;
    int nrOfElements = 10000;
    int nrOfMessages = 1000;

    pi.calculate(nrOfWorkers, nrOfElements, nrOfMessages);
  }

  static class Calculate {
  }

  static class Work {
    private final int start;
    private final int nrOfElements;

    public Work(int start, int nrOfElements) {
      this.start = start;
      this.nrOfElements = nrOfElements;
    }

    public int getStart() {
      return start;
    }

    public int getNrOfElements() {
      return nrOfElements;
    }
  }

  static class Result {
    private final double value;

    public Result(double value) {
      this.value = value;
    }

    public double getValue() {
      return value;
    }
  }

  static class PiApproximation {
    private final double pi;
    private final long duration;

    public PiApproximation(double pi, long duration) {
      this.pi = pi;
      this.duration = duration;
    }

    public double getPi() {
      return pi;
    }

    public long getDuration() {
      return duration;
    }
  }

  public static class Worker extends AbstractActor {

    private double calculatePiFor(int start, int nrOfElements) {
      double acc = 0.0;
      for (int i = start * nrOfElements; i <= ((start + 1) * nrOfElements - 1); i++) {
        acc += 4.0 * (1 - (i % 2) * 2) / (2 * i + 1);
      }
      return acc;
    }

      @Override
      public Receive createReceive() {
          ReceiveBuilder builder = ReceiveBuilder.create();
          builder.match(Work.class, w -> {
              double result = calculatePiFor(w.getStart(), w.getNrOfElements());
              getSender().tell(new Result(result), getSelf());
          });
          builder.matchAny(m -> unhandled(m));

          return builder.build();
      }
  }

  public static class Master extends AbstractActor {
      private final Router workerRouter;
      List<Routee> routeeList = new ArrayList<>();
    private final int nrOfMessages;
    private final int nrOfElements;

    private double pi;
    private int nrOfResults;
    private final long start = System.currentTimeMillis();

    private final ActorRef listener;

    public Master(
      int nrOfWorkers, 
      int nrOfMessages, 
      int nrOfElements, 
      ActorRef listener) {

      this.nrOfMessages = nrOfMessages;
      this.nrOfElements = nrOfElements;
      this.listener = listener;

        for (int i = 0; i < nrOfWorkers; i++) {
            ActorRef f = getContext().actorOf(Props.create(Worker.class));
            getContext().watch(f);
            routeeList.add(new ActorRefRoutee(f));
        }

      workerRouter = new Router(new RoundRobinRoutingLogic(), routeeList);

        for (int i = 0; i < nrOfMessages; i++) {
            workerRouter.route(new Work(i, nrOfElements), getSelf());
        }
    }

      @Override
      public Receive createReceive() {
          ReceiveBuilder builder = ReceiveBuilder.create();
           builder.match(Calculate.class, c -> {
               for (int start = 0; start < nrOfMessages; start++) {
                   workerRouter.route(new Work(start, nrOfElements), getSelf());
               }
           });
           builder.match(Result.class, r -> {
               pi += r.getValue();
               nrOfResults += 1;
               if (nrOfResults == nrOfMessages) {

                   long duration = System.currentTimeMillis() - start;
                   listener.tell(new PiApproximation(pi, duration), getSelf());

                   getContext().stop(getSelf());
               }
           });

           builder.matchAny(o -> unhandled(o));

          return builder.build();
      }
  }

  public static class Listener extends AbstractActor {

      @Override
      public Receive createReceive() {
          ReceiveBuilder builder = ReceiveBuilder.create();
          builder.match(PiApproximation.class, p -> {
              System.out.println(String.format("\n\tPi approximation: " +
                              "\t\t%s\n\tCalculation time: \t%s",
                      p.getPi(), p.getDuration()));
                        //<---------------THERE WAS getContext().system().shutdown() HERE
          });
          builder.matchAny(o -> unhandled(o));

          return builder.build();
      }
  }

  public void calculate(
    final int nrOfWorkers, 
    final int nrOfElements, 
    final int nrOfMessages) {

    ActorSystem system = ActorSystem.create("PiSystem");

    final ActorRef listener = system.actorOf(Props.create(Listener.class), "listener");

    @SuppressWarnings("serial")
    ActorRef master = system.actorOf(Props.create(Master.class, nrOfWorkers, nrOfMessages, nrOfElements, listener), "master");

    master.tell(new Calculate(), ActorRef.noSender());
  }
}

输出:

    Pi approximation:       6.2831130367262435
    Calculation time:   59
[INFO] [01/17/2018 12:08:30.955] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$b] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$b#340929109] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.956] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$a] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$a#1014207284] was not delivered. [2] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.956] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$b] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$b#340929109] was not delivered. [3] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.956] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$a] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$a#1014207284] was not delivered. [4] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.959] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$a] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$a#1014207284] was not delivered. [5] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.959] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$b] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$b#340929109] was not delivered. [6] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.959] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$b] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$b#340929109] was not delivered. [7] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.959] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$a] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$a#1014207284] was not delivered. [8] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.959] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$b] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$b#340929109] was not delivered. [9] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [01/17/2018 12:08:30.959] [PiSystem-akka.actor.default-dispatcher-5] [akka://PiSystem/user/master/$b] Message [org.example.Pi$Work] from Actor[akka://PiSystem/user/master#-1521441633] to Actor[akka://PiSystem/user/master/$b#340929109] was not delivered. [10] dead letters encountered, no more dead letters will be logged. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

Answer 1

你有

for (int i = 0; i < nrOfMessages; i++) {
  workerRouter.route(new Work(i, nrOfElements), getSelf());
}

部分在主 Actor 中重复。一次在构造函数中，一次在接收 Calculate 消息时。