raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Updated the Pi tutorial to reflect the changes in Akka 2.0. Fixes #1354

Browse source
This commit is contained in:
Henrik Engstrom 2011-12-13 22:41:38 +01:00
parent c64086fe4c
commit 4c6c316a76
3 changed files with 46 additions and 191 deletions
Download patch file Download diff file Expand all files Collapse all files

132
akka-docs/intro/code/tutorials/first/Pi.scala
View file

@ -1,132 +0,0 @@
// /**
// * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
// */
// //#imports
// package akka.tutorial.first.scala
// import akka.actor.{ Actor, ActorSystem, PoisonPill }
// import akka.routing.Routing.Broadcast
// import akka.routing.{ RoutedProps, Routing }
// import java.util.concurrent.CountDownLatch
// //#imports
// //#system
// object Pi extends App {
// val system = ActorSystem()
// calculate(nrOfWorkers = 4, nrOfElements = 10000, nrOfMessages = 10000)
// //#actors-and-messages
// // ====================
// // ===== Messages =====
// // ====================
// //#messages
// sealed trait PiMessage
// case object Calculate extends PiMessage
// case class Work(start: Int, nrOfElements: Int) extends PiMessage
// case class Result(value: Double) extends PiMessage
// //#messages
// // ==================
// // ===== Worker =====
// // ==================
// //#worker
// class Worker extends Actor {
// // define the work
// //#calculatePiFor
// def calculatePiFor(start: Int, nrOfElements: Int): Double = {
// var acc = 0.0
// for (i ← start until (start + nrOfElements))
// acc += 4.0 * (1 - (i % 2) * 2) / (2 * i + 1)
// acc
// }
// //#calculatePiFor
// def receive = {
// case Work(start, nrOfElements) ⇒ sender ! Result(calculatePiFor(start, nrOfElements)) // perform the work
// }
// }
// //#worker
// // ==================
// // ===== Master =====
// // ==================
// //#master
// class Master(nrOfWorkers: Int, nrOfMessages: Int, nrOfElements: Int, latch: CountDownLatch) extends Actor {
// var pi: Double = _
// var nrOfResults: Int = _
// var start: Long = _
// //#create-workers
// // create the workers
// val workers = Vector.fill(nrOfWorkers)(system.actorOf(Props[Worker])
// // wrap them with a load-balancing router
// val router = system.actorOf(RoutedProps().withRoundRobinRouter.withLocalConnections(workers), "pi")
// //#create-workers
// //#master-receive
// // message handler
// def receive = {
// //#handle-messages
// case Calculate ⇒
// // schedule work
// for (i ← 0 until nrOfMessages) router ! Work(i * nrOfElements, nrOfElements)
// // send a PoisonPill to all workers telling them to shut down themselves
// router ! Broadcast(PoisonPill)
// // send a PoisonPill to the router, telling him to shut himself down
// router ! PoisonPill
// case Result(value) ⇒
// // handle result from the worker
// pi += value
// nrOfResults += 1
// if (nrOfResults == nrOfMessages) self.stop()
// //#handle-messages
// }
// //#master-receive
// override def preStart() {
// start = System.currentTimeMillis
// }
// override def postStop() {
// // tell the world that the calculation is complete
// println(
// "\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis"
// .format(pi, (System.currentTimeMillis - start)))
// latch.countDown()
// }
// }
// //#master
// //#actors-and-messages
// // ==================
// // ===== Run it =====
// // ==================
// def calculate(nrOfWorkers: Int, nrOfElements: Int, nrOfMessages: Int) {
// // this latch is only plumbing to know when the calculation is completed
// val latch = new CountDownLatch(1)
// // create the master
// val master = system.actorOf(Props(new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
// // start the calculation
// master ! Calculate
// // wait for master to shut down
// latch.await()
// }
// }
// //#system

56
akka-tutorials/akka-tutorial-first/src/main/java/akka/tutorial/first/java/Pi.java
View file

@ -4,17 +4,13 @@
package akka.tutorial.first.java;
import akka.actor.Props;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
//#imports
import akka.actor.*;
import akka.routing.RoundRobinRouter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
//#imports
//#app
public class Pi {
public static void main(String[] args) throws Exception {
@ -22,9 +18,8 @@ public class Pi {
pi.calculate(4, 10000, 10000);
}
// ====================
// ===== Messages =====
// ====================
//#actors-and-messages
//#messages
static class Calculate {
}
@ -57,13 +52,12 @@ public class Pi {
return value;
}
}
//#messages
// ==================
// ===== Worker =====
// ==================
//#worker
public static class Worker extends UntypedActor {
// define the work
//#calculatePiFor
private double calculatePiFor(int start, int nrOfElements) {
double acc = 0.0;
for (int i = start * nrOfElements; i <= ((start + 1) * nrOfElements - 1); i++) {
@ -71,25 +65,22 @@ public class Pi {
}
return acc;
}
//#calculatePiFor
// message handler
public void onReceive(Object message) {
if (message instanceof Work) {
Work work = (Work) message;
// perform the work
double result = calculatePiFor(work.getStart(), work.getNrOfElements());
// reply with the result
getSender().tell(new Result(result));
} else throw new IllegalArgumentException("Unknown message [" + message + "]");
}
}
//#worker
// ==================
// ===== Master =====
// ==================
//#master
public static class Master extends UntypedActor {
private final int nrOfMessages;
private final int nrOfElements;
@ -106,28 +97,27 @@ public class Pi {
this.nrOfElements = nrOfElements;
this.latch = latch;
router = this.getContext().actorOf(new Props().withCreator(Worker.class).withRouter(new RoundRobinRouter(5)), "pi");
//#create-router
router = this.getContext().actorOf(new Props().withCreator(Worker.class).withRouter(new RoundRobinRouter(nrOfWorkers)), "pi");
//#create-router
}
// message handler
//#master-receive
public void onReceive(Object message) {
//#handle-messages
if (message instanceof Calculate) {
// schedule work
for (int start = 0; start < nrOfMessages; start++) {
router.tell(new Work(start, nrOfElements), getSelf());
}
} else if (message instanceof Result) {
// handle result from the worker
Result result = (Result) message;
pi += result.getValue();
nrOfResults += 1;
if (nrOfResults == nrOfMessages) getSelf().stop();
} else throw new IllegalArgumentException("Unknown message [" + message + "]");
//#handle-messages
}
//#master-receive
@Override
public void preStart() {
@ -136,19 +126,18 @@ public class Pi {
@Override
public void postStop() {
// tell the world that the calculation is complete
System.out.println(String.format(
"\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis",
pi, (System.currentTimeMillis() - start)));
latch.countDown();
}
}
//#master
//#actors-and-messages
// ==================
// ===== Run it =====
// ==================
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
@ -171,3 +160,4 @@ public class Pi {
system.stop();
}
}
//#app

49
akka-tutorials/akka-tutorial-first/src/main/scala/Pi.scala
View file

@ -3,49 +3,45 @@
*/
package akka.tutorial.first.scala
//#imports
import java.util.concurrent.CountDownLatch
import akka.actor._
import akka.routing._
import com.typesafe.config.{ ConfigFactory, Config }
//#imports
//#app
object Pi extends App {
// Initiate the calculation
calculate(nrOfWorkers = 4, nrOfElements = 10000, nrOfMessages = 10000)
// ====================
// ===== Messages =====
// ====================
//#actors-and-messages
//#messages
sealed trait PiMessage
case object Calculate extends PiMessage
case class Work(start: Int, nrOfElements: Int) extends PiMessage
case class Result(value: Double) extends PiMessage
//#messages
// ==================
// ===== Worker =====
// ==================
//#worker
class Worker extends Actor {
// define the work
//#calculatePiFor
def calculatePiFor(start: Int, nrOfElements: Int): Double = {
var acc = 0.0
for (i ← start until (start + nrOfElements))
acc += 4.0 * (1 - (i % 2) * 2) / (2 * i + 1)
acc
}
//#calculatePiFor
def receive = {
case Work(start, nrOfElements) ⇒
sender ! Result(calculatePiFor(start, nrOfElements)) // perform the work
}
}
//#worker
// ==================
// ===== Master =====
// ==================
//#master
class Master(nrOfWorkers: Int, nrOfMessages: Int, nrOfElements: Int, latch: CountDownLatch)
extends Actor {
@ -53,40 +49,40 @@ object Pi extends App {
var nrOfResults: Int = _
var start: Long = _
// create a round robin router for the workers
val router = context.actorOf(Props(new Worker).withRouter(RoundRobinRouter(nrOfInstances = 5)), "pi")
//#create-router
val router = context.actorOf(Props(new Worker).withRouter(RoundRobinRouter(nrOfInstances = nrOfWorkers)), "pi")
//#create-router
// message handler
//#master-receive
def receive = {
//#handle-messages
case Calculate ⇒
// schedule work
for (i ← 0 until nrOfMessages) router ! Work(i * nrOfElements, nrOfElements)
case Result(value) ⇒
// handle result from the worker
pi += value
nrOfResults += 1
// Stop this actor and all its supervised children
// Stops this actor and all its supervised children
if (nrOfResults == nrOfMessages) self.stop()
//#handle-messages
}
//#master-receive
override def preStart() {
start = System.currentTimeMillis
}
override def postStop() {
// tell the world that the calculation is complete
println(
"\n\tPi estimate: \t\t%s\n\tCalculation time: \t%s millis"
.format(pi, (System.currentTimeMillis - start)))
latch.countDown()
}
}
//#master
//#actors-and-messages
// ==================
// ===== Run it =====
// ==================
def calculate(nrOfWorkers: Int, nrOfElements: Int, nrOfMessages: Int) {
// Create an Akka system
val system = ActorSystem("PiSystem")
// this latch is only plumbing to know when the calculation is completed
@ -105,3 +101,4 @@ object Pi extends App {
system.stop()
}
}
//#app