Improved java version of pi example also. See #1729
This commit is contained in:
Patrik Nordwall
parent
c64b73004a
commit
fa41cea897
5 changed files with 239 additions and 200 deletions
|
|
@ -5,172 +5,193 @@
|
|||
package akka.tutorial.first.java;
|
||||
|
||||
//#imports
|
||||
import akka.actor.Props;
|
||||
|
||||
import akka.actor.ActorRef;
|
||||
import akka.actor.ActorSystem;
|
||||
import akka.actor.InternalActorRef;
|
||||
import akka.actor.Props;
|
||||
import akka.actor.UntypedActor;
|
||||
import akka.actor.UntypedActorFactory;
|
||||
import akka.japi.Creator;
|
||||
import akka.routing.*;
|
||||
import akka.util.Timeout;
|
||||
import akka.routing.RoundRobinRouter;
|
||||
import akka.util.Duration;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import java.util.LinkedList;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
//#imports
|
||||
|
||||
//#app
|
||||
public class Pi {
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
Pi pi = new Pi();
|
||||
pi.calculate(4, 10000, 10000);
|
||||
public static void main(String[] args) {
|
||||
Pi pi = new Pi();
|
||||
pi.calculate(4, 10000, 10000);
|
||||
}
|
||||
|
||||
//#actors-and-messages
|
||||
//#messages
|
||||
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;
|
||||
}
|
||||
|
||||
//#actors-and-messages
|
||||
//#messages
|
||||
static class Calculate {
|
||||
public int getStart() {
|
||||
return start;
|
||||
}
|
||||
|
||||
static class Work {
|
||||
private final int start;
|
||||
private final int nrOfElements;
|
||||
public int getNrOfElements() {
|
||||
return nrOfElements;
|
||||
}
|
||||
}
|
||||
|
||||
public Work(int start, int nrOfElements) {
|
||||
this.start = start;
|
||||
this.nrOfElements = nrOfElements;
|
||||
}
|
||||
static class Result {
|
||||
private final double value;
|
||||
|
||||
public int getStart() {
|
||||
return start;
|
||||
}
|
||||
|
||||
public int getNrOfElements() {
|
||||
return nrOfElements;
|
||||
}
|
||||
public Result(double value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
static class Result {
|
||||
private final double value;
|
||||
|
||||
public Result(double value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
public double getValue() {
|
||||
return value;
|
||||
}
|
||||
public double getValue() {
|
||||
return value;
|
||||
}
|
||||
//#messages
|
||||
}
|
||||
|
||||
//#worker
|
||||
public static class Worker extends UntypedActor {
|
||||
static class PiApproximation {
|
||||
private final double pi;
|
||||
private final Duration duration;
|
||||
|
||||
//#calculatePiFor
|
||||
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;
|
||||
}
|
||||
//#calculatePiFor
|
||||
|
||||
public void onReceive(Object message) {
|
||||
if (message instanceof Work) {
|
||||
Work work = (Work) message;
|
||||
double result = calculatePiFor(work.getStart(), work.getNrOfElements());
|
||||
getSender().tell(new Result(result));
|
||||
} else {
|
||||
throw new IllegalArgumentException("Unknown message [" + message + "]");
|
||||
}
|
||||
}
|
||||
public PiApproximation(double pi, Duration duration) {
|
||||
this.pi = pi;
|
||||
this.duration = duration;
|
||||
}
|
||||
//#worker
|
||||
|
||||
//#master
|
||||
public static class Master extends UntypedActor {
|
||||
private final int nrOfMessages;
|
||||
private final int nrOfElements;
|
||||
private final CountDownLatch latch;
|
||||
|
||||
private double pi;
|
||||
private int nrOfResults;
|
||||
private long start;
|
||||
|
||||
private ActorRef router;
|
||||
|
||||
public Master(final int nrOfWorkers, int nrOfMessages,
|
||||
int nrOfElements, CountDownLatch latch) {
|
||||
this.nrOfMessages = nrOfMessages;
|
||||
this.nrOfElements = nrOfElements;
|
||||
this.latch = latch;
|
||||
|
||||
//#create-router
|
||||
router = this.getContext().actorOf(
|
||||
new Props(Worker.class).withRouter(new RoundRobinRouter(nrOfWorkers)),
|
||||
"pi");
|
||||
//#create-router
|
||||
}
|
||||
|
||||
//#master-receive
|
||||
public void onReceive(Object message) {
|
||||
//#handle-messages
|
||||
if (message instanceof Calculate) {
|
||||
for (int start = 0; start < nrOfMessages; start++) {
|
||||
router.tell(new Work(start, nrOfElements), getSelf());
|
||||
}
|
||||
} else if (message instanceof Result) {
|
||||
Result result = (Result) message;
|
||||
pi += result.getValue();
|
||||
nrOfResults += 1;
|
||||
if (nrOfResults == nrOfMessages) getContext().stop(getSelf());
|
||||
} else throw new IllegalArgumentException("Unknown message [" + message + "]");
|
||||
//#handle-messages
|
||||
}
|
||||
//#master-receive
|
||||
|
||||
@Override
|
||||
public void preStart() {
|
||||
start = System.currentTimeMillis();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void postStop() {
|
||||
System.out.println(String.format(
|
||||
"\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis",
|
||||
pi, (System.currentTimeMillis() - start)));
|
||||
latch.countDown();
|
||||
}
|
||||
public double getPi() {
|
||||
return pi;
|
||||
}
|
||||
//#master
|
||||
//#actors-and-messages
|
||||
|
||||
public void calculate(final int nrOfWorkers,
|
||||
final int nrOfElements,
|
||||
final int nrOfMessages)
|
||||
throws Exception {
|
||||
// Create an Akka system
|
||||
final ActorSystem system = ActorSystem.create();
|
||||
|
||||
// this latch is only plumbing to know when the calculation is completed
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
|
||||
// create the master
|
||||
ActorRef master = system.actorOf(new Props(new UntypedActorFactory() {
|
||||
public UntypedActor create() {
|
||||
return new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch);
|
||||
}
|
||||
}));
|
||||
|
||||
// start the calculation
|
||||
master.tell(new Calculate());
|
||||
|
||||
// wait for master to shut down
|
||||
latch.await();
|
||||
|
||||
// Shut down the system
|
||||
system.shutdown();
|
||||
public Duration getDuration() {
|
||||
return duration;
|
||||
}
|
||||
}
|
||||
|
||||
//#messages
|
||||
|
||||
//#worker
|
||||
public static class Worker extends UntypedActor {
|
||||
|
||||
//#calculatePiFor
|
||||
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;
|
||||
}
|
||||
|
||||
//#calculatePiFor
|
||||
|
||||
public void onReceive(Object message) {
|
||||
if (message instanceof Work) {
|
||||
Work work = (Work) message;
|
||||
double result = calculatePiFor(work.getStart(), work.getNrOfElements());
|
||||
getSender().tell(new Result(result), getSelf());
|
||||
} else {
|
||||
unhandled(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//#worker
|
||||
|
||||
//#master
|
||||
public static class Master extends UntypedActor {
|
||||
private final int nrOfMessages;
|
||||
private final int nrOfElements;
|
||||
|
||||
private double pi;
|
||||
private int nrOfResults;
|
||||
private final long start = System.currentTimeMillis();
|
||||
|
||||
private final ActorRef listener;
|
||||
private final ActorRef workerRouter;
|
||||
|
||||
public Master(final int nrOfWorkers, int nrOfMessages, int nrOfElements, ActorRef listener) {
|
||||
this.nrOfMessages = nrOfMessages;
|
||||
this.nrOfElements = nrOfElements;
|
||||
this.listener = listener;
|
||||
|
||||
//#create-router
|
||||
workerRouter = this.getContext().actorOf(new Props(Worker.class).withRouter(new RoundRobinRouter(nrOfWorkers)),
|
||||
"workerRouter");
|
||||
//#create-router
|
||||
}
|
||||
|
||||
//#master-receive
|
||||
public void onReceive(Object message) {
|
||||
//#handle-messages
|
||||
if (message instanceof Calculate) {
|
||||
for (int start = 0; start < nrOfMessages; start++) {
|
||||
workerRouter.tell(new Work(start, nrOfElements), getSelf());
|
||||
}
|
||||
} else if (message instanceof Result) {
|
||||
Result result = (Result) message;
|
||||
pi += result.getValue();
|
||||
nrOfResults += 1;
|
||||
if (nrOfResults == nrOfMessages) {
|
||||
// Send the result to the listener
|
||||
Duration duration = Duration.create(System.currentTimeMillis() - start, TimeUnit.MILLISECONDS);
|
||||
listener.tell(new PiApproximation(pi, duration), getSelf());
|
||||
// Stops this actor and all its supervised children
|
||||
getContext().stop(getSelf());
|
||||
}
|
||||
} else {
|
||||
unhandled(message);
|
||||
}
|
||||
//#handle-messages
|
||||
}
|
||||
//#master-receive
|
||||
}
|
||||
|
||||
//#master
|
||||
|
||||
//#result-listener
|
||||
public static class Listener extends UntypedActor {
|
||||
public void onReceive(Object message) {
|
||||
if (message instanceof PiApproximation) {
|
||||
PiApproximation approximation = (PiApproximation) message;
|
||||
System.out.println(String.format("\n\tPi approximation: \t\t%s\n\tCalculation time: \t%s",
|
||||
approximation.getPi(), approximation.getDuration()));
|
||||
getContext().system().shutdown();
|
||||
} else {
|
||||
unhandled(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
//#result-listener
|
||||
//#actors-and-messages
|
||||
|
||||
public void calculate(final int nrOfWorkers, final int nrOfElements, final int nrOfMessages) {
|
||||
// Create an Akka system
|
||||
ActorSystem system = ActorSystem.create("PiSystem");
|
||||
|
||||
// create the result listener, which will print the result and shutdown the system
|
||||
final ActorRef listener = system.actorOf(new Props(Listener.class));
|
||||
|
||||
// create the master
|
||||
ActorRef master = system.actorOf(new Props(new UntypedActorFactory() {
|
||||
public UntypedActor create() {
|
||||
return new Master(nrOfWorkers, nrOfMessages, nrOfElements, listener);
|
||||
}
|
||||
}), "master");
|
||||
|
||||
// start the calculation
|
||||
master.tell(new Calculate());
|
||||
|
||||
}
|
||||
}
|
||||
//#app
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ package akka.tutorial.first.scala
|
|||
|
||||
//#imports
|
||||
import akka.actor._
|
||||
import akka.routing._
|
||||
import akka.routing.RoundRobinRouter
|
||||
import akka.util.Duration
|
||||
import akka.util.duration._
|
||||
//#imports
|
||||
|
|
@ -21,7 +21,7 @@ object Pi extends App {
|
|||
case object Calculate extends PiMessage
|
||||
case class Work(start: Int, nrOfElements: Int) extends PiMessage
|
||||
case class Result(value: Double) extends PiMessage
|
||||
case class PiEstimate(pi: Double, duration: Duration)
|
||||
case class PiApproximation(pi: Double, duration: Duration)
|
||||
//#messages
|
||||
|
||||
//#worker
|
||||
|
|
@ -52,21 +52,21 @@ object Pi extends App {
|
|||
val start: Long = System.currentTimeMillis
|
||||
|
||||
//#create-router
|
||||
val router = context.actorOf(
|
||||
Props[Worker].withRouter(RoundRobinRouter(nrOfWorkers)), name = "workers")
|
||||
val workerRouter = context.actorOf(
|
||||
Props[Worker].withRouter(RoundRobinRouter(nrOfWorkers)), name = "workerRouter")
|
||||
//#create-router
|
||||
|
||||
//#master-receive
|
||||
def receive = {
|
||||
//#handle-messages
|
||||
case Calculate ⇒
|
||||
for (i ← 0 until nrOfMessages) router ! Work(i * nrOfElements, nrOfElements)
|
||||
for (i ← 0 until nrOfMessages) workerRouter ! Work(i * nrOfElements, nrOfElements)
|
||||
case Result(value) ⇒
|
||||
pi += value
|
||||
nrOfResults += 1
|
||||
if (nrOfResults == nrOfMessages) {
|
||||
// Send the result to the listener
|
||||
listener ! PiEstimate(pi, duration = (System.currentTimeMillis - start).millis)
|
||||
listener ! PiApproximation(pi, duration = (System.currentTimeMillis - start).millis)
|
||||
// Stops this actor and all its supervised children
|
||||
context.stop(self)
|
||||
}
|
||||
|
|
@ -80,8 +80,8 @@ object Pi extends App {
|
|||
//#result-listener
|
||||
class Listener extends Actor {
|
||||
def receive = {
|
||||
case PiEstimate(pi, duration) ⇒
|
||||
println("\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s"
|
||||
case PiApproximation(pi, duration) ⇒
|
||||
println("\n\tPi approximation: \t\t%s\n\tCalculation time: \t%s"
|
||||
.format(pi, duration))
|
||||
context.system.shutdown()
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue