Updated samples and tutorial to Akka 2.0. Added projects to SBT project file. Fixes #1278
This commit is contained in:
Henrik Engstrom
parent
c0d3c523e2
commit
823a68ac0f
33 changed files with 1291 additions and 1209 deletions
|
|
@ -1,182 +1,184 @@
|
|||
// *
|
||||
// * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
|
||||
/**
|
||||
* Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
|
||||
package akka.tutorial.first.java;
|
||||
|
||||
// package akka.tutorial.first.java;
|
||||
import akka.actor.ActorRef;
|
||||
import akka.actor.ActorSystem;
|
||||
import akka.actor.UntypedActor;
|
||||
import akka.actor.UntypedActorFactory;
|
||||
import akka.japi.Creator;
|
||||
import akka.routing.*;
|
||||
|
||||
// import static akka.actor.Actors.poisonPill;
|
||||
// import static java.util.Arrays.asList;
|
||||
import java.util.LinkedList;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
|
||||
// import akka.actor.ActorRef;
|
||||
// import akka.actor.Actors;
|
||||
// import akka.actor.ActorSystem;
|
||||
// import akka.actor.UntypedActor;
|
||||
// import akka.actor.UntypedActorFactory;
|
||||
// import akka.routing.RoutedProps;
|
||||
// import akka.routing.RouterType;
|
||||
// import akka.routing.LocalConnectionManager;
|
||||
// import akka.routing.Routing;
|
||||
// import akka.routing.Routing.Broadcast;
|
||||
// import scala.collection.JavaConversions;
|
||||
public class Pi {
|
||||
|
||||
// import java.util.LinkedList;
|
||||
// import java.util.concurrent.CountDownLatch;
|
||||
private static final ActorSystem system = ActorSystem.apply();
|
||||
|
||||
// public class Pi {
|
||||
public static void main(String[] args) throws Exception {
|
||||
Pi pi = new Pi();
|
||||
pi.calculate(4, 10000, 10000);
|
||||
}
|
||||
|
||||
// private static final ActorSystem system = new ActorSystem();
|
||||
// ====================
|
||||
// ===== Messages =====
|
||||
// ====================
|
||||
static class Calculate {
|
||||
}
|
||||
|
||||
// public static void main(String[] args) throws Exception {
|
||||
// Pi pi = new Pi();
|
||||
// pi.calculate(4, 10000, 10000);
|
||||
// }
|
||||
static class Work {
|
||||
private final int start;
|
||||
private final int nrOfElements;
|
||||
|
||||
// // ====================
|
||||
// // ===== Messages =====
|
||||
// // ====================
|
||||
// static class Calculate {}
|
||||
public Work(int start, int nrOfElements) {
|
||||
this.start = start;
|
||||
this.nrOfElements = nrOfElements;
|
||||
}
|
||||
|
||||
// static class Work {
|
||||
// private final int start;
|
||||
// private final int nrOfElements;
|
||||
public int getStart() {
|
||||
return start;
|
||||
}
|
||||
|
||||
// public Work(int start, int nrOfElements) {
|
||||
// this.start = start;
|
||||
// this.nrOfElements = nrOfElements;
|
||||
// }
|
||||
public int getNrOfElements() {
|
||||
return nrOfElements;
|
||||
}
|
||||
}
|
||||
|
||||
// public int getStart() { return start; }
|
||||
// public int getNrOfElements() { return nrOfElements; }
|
||||
// }
|
||||
static class Result {
|
||||
private final double value;
|
||||
|
||||
// static class Result {
|
||||
// private final double value;
|
||||
public Result(double value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
// public Result(double value) {
|
||||
// this.value = value;
|
||||
// }
|
||||
public double getValue() {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
|
||||
// public double getValue() { return value; }
|
||||
// }
|
||||
// ==================
|
||||
// ===== Worker =====
|
||||
// ==================
|
||||
public static class Worker extends UntypedActor {
|
||||
|
||||
// // ==================
|
||||
// // ===== Worker =====
|
||||
// // ==================
|
||||
// static class Worker extends UntypedActor {
|
||||
// define the work
|
||||
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;
|
||||
}
|
||||
|
||||
// // define the work
|
||||
// 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;
|
||||
// }
|
||||
// message handler
|
||||
public void onReceive(Object message) {
|
||||
if (message instanceof Work) {
|
||||
Work work = (Work) message;
|
||||
|
||||
// // 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());
|
||||
|
||||
// // perform the work
|
||||
// double result = calculatePiFor(work.getStart(), work.getNrOfElements());
|
||||
// reply with the result
|
||||
getSender().tell(new Result(result));
|
||||
|
||||
// // reply with the result
|
||||
// getSender().tell(new Result(result));
|
||||
} else throw new IllegalArgumentException("Unknown message [" + message + "]");
|
||||
}
|
||||
}
|
||||
|
||||
// } else throw new IllegalArgumentException("Unknown message [" + message + "]");
|
||||
// }
|
||||
// }
|
||||
// ==================
|
||||
// ===== Master =====
|
||||
// ==================
|
||||
public static class Master extends UntypedActor {
|
||||
private final int nrOfMessages;
|
||||
private final int nrOfElements;
|
||||
private final CountDownLatch latch;
|
||||
|
||||
// // ==================
|
||||
// // ===== Master =====
|
||||
// // ==================
|
||||
// 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 double pi;
|
||||
// private int nrOfResults;
|
||||
// private long start;
|
||||
private ActorRef router;
|
||||
|
||||
// private ActorRef router;
|
||||
public Master(final int nrOfWorkers, int nrOfMessages, int nrOfElements, CountDownLatch latch) {
|
||||
this.nrOfMessages = nrOfMessages;
|
||||
this.nrOfElements = nrOfElements;
|
||||
this.latch = latch;
|
||||
Creator<Router> routerCreator = new Creator<Router>() {
|
||||
public Router create() {
|
||||
return new RoundRobinRouter();
|
||||
}
|
||||
};
|
||||
LinkedList<ActorRef> actors = new LinkedList<ActorRef>() {
|
||||
{
|
||||
for (int i = 0; i < nrOfWorkers; i++) add(system.actorOf(Worker.class));
|
||||
}
|
||||
};
|
||||
RoutedProps props = new RoutedProps(routerCreator, new LocalConnectionManager(actors), new akka.actor.Timeout(-1), true);
|
||||
router = new RoutedActorRef(system, props, getSelf(), "pi");
|
||||
}
|
||||
|
||||
// public Master(int nrOfWorkers, int nrOfMessages, int nrOfElements, CountDownLatch latch) {
|
||||
// this.nrOfMessages = nrOfMessages;
|
||||
// this.nrOfElements = nrOfElements;
|
||||
// this.latch = latch;
|
||||
// message handler
|
||||
public void onReceive(Object message) {
|
||||
|
||||
// LinkedList<ActorRef> workers = new LinkedList<ActorRef>();
|
||||
// for (int i = 0; i < nrOfWorkers; i++) {
|
||||
// ActorRef worker = system.actorOf(Worker.class);
|
||||
// workers.add(worker);
|
||||
// }
|
||||
if (message instanceof Calculate) {
|
||||
|
||||
// router = system.actorOf(new RoutedProps().withRoundRobinRouter().withLocalConnections(workers), "pi");
|
||||
// }
|
||||
// schedule work
|
||||
for (int start = 0; start < nrOfMessages; start++) {
|
||||
router.tell(new Work(start, nrOfElements), getSelf());
|
||||
}
|
||||
|
||||
// // message handler
|
||||
// public void onReceive(Object message) {
|
||||
} else if (message instanceof Result) {
|
||||
|
||||
// if (message instanceof Calculate) {
|
||||
// // schedule work
|
||||
// for (int start = 0; start < nrOfMessages; start++) {
|
||||
// router.tell(new Work(start, nrOfElements), getSelf());
|
||||
// }
|
||||
// handle result from the worker
|
||||
Result result = (Result) message;
|
||||
pi += result.getValue();
|
||||
nrOfResults += 1;
|
||||
if (nrOfResults == nrOfMessages) getSelf().stop();
|
||||
|
||||
// // send a PoisonPill to all workers telling them to shut down themselves
|
||||
// router.tell(new Broadcast(poisonPill()));
|
||||
} else throw new IllegalArgumentException("Unknown message [" + message + "]");
|
||||
}
|
||||
|
||||
// // send a PoisonPill to the router, telling him to shut himself down
|
||||
// router.tell(poisonPill());
|
||||
@Override
|
||||
public void preStart() {
|
||||
start = System.currentTimeMillis();
|
||||
}
|
||||
|
||||
// } else if (message instanceof Result) {
|
||||
@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();
|
||||
}
|
||||
}
|
||||
|
||||
// // handle result from the worker
|
||||
// Result result = (Result) message;
|
||||
// pi += result.getValue();
|
||||
// nrOfResults += 1;
|
||||
// if (nrOfResults == nrOfMessages) getSelf().stop();
|
||||
// ==================
|
||||
// ===== Run it =====
|
||||
// ==================
|
||||
public void calculate(final int nrOfWorkers, final int nrOfElements, final int nrOfMessages)
|
||||
throws Exception {
|
||||
|
||||
// } else throw new IllegalArgumentException("Unknown message [" + message + "]");
|
||||
// }
|
||||
// this latch is only plumbing to know when the calculation is completed
|
||||
final CountDownLatch latch = new CountDownLatch(1);
|
||||
|
||||
// @Override
|
||||
// public void preStart() {
|
||||
// start = System.currentTimeMillis();
|
||||
// }
|
||||
// create the master
|
||||
ActorRef master = system.actorOf(new UntypedActorFactory() {
|
||||
public UntypedActor create() {
|
||||
return new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch);
|
||||
}
|
||||
});
|
||||
|
||||
// @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();
|
||||
// }
|
||||
// }
|
||||
// start the calculation
|
||||
master.tell(new Calculate());
|
||||
|
||||
// // ==================
|
||||
// // ===== Run it =====
|
||||
// // ==================
|
||||
// public void calculate(final int nrOfWorkers, final int nrOfElements, final int nrOfMessages)
|
||||
// throws Exception {
|
||||
// wait for master to shut down
|
||||
latch.await();
|
||||
|
||||
// // 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 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.stop();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,113 +1,114 @@
|
|||
// /**
|
||||
// * Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
|
||||
// */
|
||||
/**
|
||||
* Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
package akka.tutorial.first.scala
|
||||
|
||||
// package akka.tutorial.first.scala
|
||||
import java.util.concurrent.CountDownLatch
|
||||
import akka.routing.{ RoutedActorRef, LocalConnectionManager, RoundRobinRouter, RoutedProps }
|
||||
import akka.actor.{ ActorSystemImpl, Actor, ActorSystem }
|
||||
|
||||
// import akka.actor.{ Actor, PoisonPill, ActorSystem }
|
||||
// import Actor._
|
||||
// import java.util.concurrent.CountDownLatch
|
||||
// import akka.routing.Routing.Broadcast
|
||||
// import akka.routing.{ RoutedProps, Routing }
|
||||
object Pi extends App {
|
||||
|
||||
// object Pi extends App {
|
||||
val system = ActorSystem()
|
||||
|
||||
// val system = ActorSystem()
|
||||
// Initiate the calculation
|
||||
calculate(nrOfWorkers = 4, nrOfElements = 10000, nrOfMessages = 10000)
|
||||
|
||||
// calculate(nrOfWorkers = 4, nrOfElements = 10000, nrOfMessages = 10000)
|
||||
// ====================
|
||||
// ===== Messages =====
|
||||
// ====================
|
||||
sealed trait PiMessage
|
||||
|
||||
// // ====================
|
||||
// // ===== Messages =====
|
||||
// // ====================
|
||||
// sealed trait PiMessage
|
||||
case object Calculate extends PiMessage
|
||||
|
||||
// case object Calculate extends PiMessage
|
||||
case class Work(start: Int, nrOfElements: Int) extends PiMessage
|
||||
|
||||
// case class Work(start: Int, nrOfElements: Int) extends PiMessage
|
||||
case class Result(value: Double) extends PiMessage
|
||||
|
||||
// case class Result(value: Double) extends PiMessage
|
||||
// ==================
|
||||
// ===== Worker =====
|
||||
// ==================
|
||||
class Worker extends Actor {
|
||||
|
||||
// // ==================
|
||||
// // ===== Worker =====
|
||||
// // ==================
|
||||
// class Worker extends Actor {
|
||||
// define the work
|
||||
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
|
||||
}
|
||||
|
||||
// // define the work
|
||||
// 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
|
||||
// }
|
||||
def receive = {
|
||||
case Work(start, nrOfElements) ⇒ sender ! Result(calculatePiFor(start, nrOfElements)) // perform the work
|
||||
}
|
||||
}
|
||||
|
||||
// def receive = {
|
||||
// case Work(start, nrOfElements) ⇒ sender ! Result(calculatePiFor(start, nrOfElements)) // perform the work
|
||||
// }
|
||||
// }
|
||||
// ==================
|
||||
// ===== Master =====
|
||||
// ==================
|
||||
class Master(nrOfWorkers: Int, nrOfMessages: Int, nrOfElements: Int, latch: CountDownLatch)
|
||||
extends Actor {
|
||||
|
||||
// // ==================
|
||||
// // ===== Master =====
|
||||
// // ==================
|
||||
// class Master(nrOfWorkers: Int, nrOfMessages: Int, nrOfElements: Int, latch: CountDownLatch)
|
||||
// extends Actor {
|
||||
var pi: Double = _
|
||||
var nrOfResults: Int = _
|
||||
var start: Long = _
|
||||
|
||||
// var pi: Double = _
|
||||
// var nrOfResults: Int = _
|
||||
// var start: Long = _
|
||||
// create the workers
|
||||
val workers = Vector.fill(nrOfWorkers)(system.actorOf[Worker])
|
||||
|
||||
// // create the workers
|
||||
// val workers = Vector.fill(nrOfWorkers)(system.actorOf[Worker])
|
||||
// wrap them with a load-balancing router
|
||||
val props = RoutedProps(routerFactory = () ⇒ new RoundRobinRouter, connectionManager = new LocalConnectionManager(workers))
|
||||
val router = new RoutedActorRef(system, props, self, "pi")
|
||||
|
||||
// // wrap them with a load-balancing router
|
||||
// val router = system.actorOf(RoutedProps().withRoundRobinRouter.withLocalConnections(workers), "pi")
|
||||
// message handler
|
||||
def receive = {
|
||||
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
|
||||
|
||||
// // message handler
|
||||
// def receive = {
|
||||
// case Calculate ⇒
|
||||
// // schedule work
|
||||
// for (i ← 0 until nrOfMessages) router ! Work(i * nrOfElements, nrOfElements)
|
||||
// Stop this actor and all its supervised children
|
||||
if (nrOfResults == nrOfMessages) self.stop()
|
||||
}
|
||||
|
||||
// // send a PoisonPill to all workers telling them to shut down themselves
|
||||
// router ! Broadcast(PoisonPill)
|
||||
override def preStart() {
|
||||
start = System.currentTimeMillis
|
||||
}
|
||||
|
||||
// // send a PoisonPill to the router, telling him to shut himself down
|
||||
// router ! PoisonPill
|
||||
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()
|
||||
}
|
||||
}
|
||||
|
||||
// case Result(value) ⇒
|
||||
// // handle result from the worker
|
||||
// pi += value
|
||||
// nrOfResults += 1
|
||||
// if (nrOfResults == nrOfMessages) self.stop()
|
||||
// }
|
||||
object Master {
|
||||
val impl = system.asInstanceOf[ActorSystemImpl]
|
||||
}
|
||||
|
||||
// override def preStart() {
|
||||
// start = System.currentTimeMillis
|
||||
// }
|
||||
// ==================
|
||||
// ===== Run it =====
|
||||
// ==================
|
||||
def calculate(nrOfWorkers: Int, nrOfElements: Int, nrOfMessages: Int) {
|
||||
|
||||
// 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()
|
||||
// }
|
||||
// }
|
||||
// this latch is only plumbing to know when the calculation is completed
|
||||
val latch = new CountDownLatch(1)
|
||||
|
||||
// // ==================
|
||||
// // ===== Run it =====
|
||||
// // ==================
|
||||
// def calculate(nrOfWorkers: Int, nrOfElements: Int, nrOfMessages: Int) {
|
||||
// create the master
|
||||
val master = system.actorOf(new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
|
||||
|
||||
// // this latch is only plumbing to know when the calculation is completed
|
||||
// val latch = new CountDownLatch(1)
|
||||
// start the calculation
|
||||
master ! Calculate
|
||||
|
||||
// // create the master
|
||||
// val master = system.actorOf(new Master(nrOfWorkers, nrOfMessages, nrOfElements, latch))
|
||||
// wait for master to shut down
|
||||
latch.await()
|
||||
|
||||
// // start the calculation
|
||||
// master ! Calculate
|
||||
|
||||
// // wait for master to shut down
|
||||
// latch.await()
|
||||
// }
|
||||
// }
|
||||
// Shut down the system
|
||||
system.stop()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,26 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
package akka.tutorial.first.scala
|
||||
|
||||
import org.junit.runner.RunWith
|
||||
import org.scalatest.matchers.MustMatchers
|
||||
import org.scalatest.WordSpec
|
||||
import akka.testkit.TestActorRef
|
||||
import akka.tutorial.first.scala.Pi.Worker
|
||||
import akka.actor.ActorSystem
|
||||
|
||||
@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
|
||||
class WorkerSpec extends WordSpec with MustMatchers {
|
||||
|
||||
implicit def system = ActorSystem()
|
||||
|
||||
"Worker" must {
|
||||
"calculate pi correctly" in {
|
||||
val testActor = TestActorRef[Worker]
|
||||
val actor = testActor.underlyingActor
|
||||
actor.calculatePiFor(0, 0) must equal(0.0)
|
||||
actor.calculatePiFor(1, 1) must equal(-1.3333333333333333)
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue