public class MonActeur extends UntypedActor {
    public void onReceive(Object message) throws Exception {
        if (message instanceof String) {
            System.out.println("Message recu" + msg);
        }
      }
}

final ActorSystem systeme = ActorSystem.create("MonApplication");

final ActorRef monActeur = systeme.actorOf(Props.create(MyActor.class));

public class Master extends UntypedActor {

	/**
	 * La liste de nombres en entree
	 */
	private final List<Integer> inputNumbers;
	
	/**
	 * La liste de nombres transformés
	 */
	private final List<Integer> transformedNumbers;	

	/**
	 * Le router
	 */
private final Router router ;
	private final long start = System.currentTimeMillis();
	
	/**
	 * Reference vers la future pour lui renvoyer le resultat
	 */
	private ActorRef initialSender = null;

	public Master(int nbWorkers, List<Integer> numbers) {
		transformedNumbers = new ArrayList<Integer>();
		this.inputNumbers = numbers;
		router = createRouter(nbWorkers);
		//System.out.println("master and router created");
	}

/**
	 * Cree un routeur contenant le pool de workers
	 * @param nbWorkers
	 * @return
	 */
	private Router createRouter (int nbWorkers) {
		Router router = null;
		final List<Routee> routees = new ArrayList<Routee>();
		 for (int i = 0; i < nbWorkers; i++) {
			 final ActorRef workerRouter = getContext().actorOf(Props.create(Worker.class));
			 getContext().watch(workerRouter);
			 routees.add(new ActorRefRoutee(workerRouter));
		 }
		 router = new Router(new RoundRobinRoutingLogic(), routees);
		 return router;
	}

@Override
	public void onReceive(Object message) {
		if (message instanceof com.poc.acteur.StartMessage) {
			initialSender = getSender();
			//System.out.println("master received a start event");
			for (Integer number : inputNumbers) {
				// le routeur dispatche les differents elements de la liste au workers
				router.route(number, getSelf());
			}
			
			//reception du resultat du traitement
		} else if (message instanceof ResultMessage) {
			//System.out.println("master received result event");
			ResultMessage result = (ResultMessage) message;
			transformedNumbers.add(result.getValue());
			
			//on detecte quand on a fini le traitement
			if (transformedNumbers.size() == inputNumbers.size()) {
				
				//calcule la duree du traitement
				Duration duration = Duration.create(System.currentTimeMillis()
						- start, TimeUnit.MILLISECONDS);
				System.out.println(transformedNumbers.size()
						+ " numbers computed in " + duration);
				
				//on renvoie le resultat a la future
				initialSender.tell(transformedNumbers, getSelf());
				
				// arrete le master et ses workers
				getContext().stop(getSelf());
				getContext().system().shutdown();
			}
		}
	}

package com.poc.acteur;

import java.io.Serializable;

public final class StartMessage implements Serializable{

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

}

package com.poc.acteur;

import java.io.Serializable;

public final class ResultMessage implements Serializable{
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	public ResultMessage(final int value) {
		this.value = value;
	}

	private final  int value;

	public int getValue() {
		return value;
	}

public class Worker extends UntypedActor {
	
	private static final int TIMER = 4000;

	@Override
	public void onReceive(Object message) {
		//System.out.println("worker received an event");
		if (message instanceof Integer) {
			int number = (Integer) message;
			int transformedNumber = tranform(number);
			// send result to master
			getSender().tell(new ResultMessage(transformedNumber), self());
		}
	}

	private int tranform(int number) {
		try {
			Thread.sleep(TIMER);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return number * 2;
	}
}

package com.poc.acteur;

import java.util.Arrays;
import java.util.List;

import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.util.Timeout;

/**
 * Programme principal
 *
 */
public class App 
{
	//taille du pool des acteurs de type worker 
	public static final int POOL_WORKER = 8;
    public static void main( String[] args ) throws Exception
    {
    	//liste d'entier a traiter
    	List<Integer> numbers = Arrays.asList(1,2,3,4,5,6,7,8);  
    	ActorSystem system = ActorSystem.create("mySystem");  
    	
    	//creation du master
    	ActorRef master = system.actorOf(Props.create(Master.class,POOL_WORKER,numbers));
    	
    	Timeout timeout = new Timeout(Duration.create("30 seconds").toMillis());
    	
		// Creation de la future, Acteur implicite auquel notre master pourra repondre
		Future future = akka.pattern.Patterns
				.ask(master, new StartMessage(), timeout);
    	
		List <Integer> listeResultat = (List<Integer>) Await.result(future,
				timeout.duration());
    	System.out.println("resultat= " + listeResultat);
		System.out.println("nb threads = " + Thread.activeCount());
		system.shutdown();
    	
    }
}

public class WorkerActor extends UntypedActor {
	LoggingAdapter log = Logging.getLogger(getContext().system(), this);
	private int state = 4;

	…

	public void onReceive(Object o) throws Exception {
		  if (o instanceof IntegerMessage) {
			IntegerMessage message = (IntegerMessage) o;
			//recupere l'entier à traiter
			int value = message.getNombre().intValue(); 
			state = state/value;
		} else if (o instanceof ResultMessageRequest) {
			getSender().tell(state, ActorRef.noSender());
		} 
	}
	…

}

package com.poc.acteur.supervisor;

import static akka.actor.SupervisorStrategy.escalate;
import static akka.actor.SupervisorStrategy.restart;
import static akka.actor.SupervisorStrategy.resume;
import scala.concurrent.duration.Duration;
import akka.actor.ActorRef;
import akka.actor.OneForOneStrategy;
import akka.actor.Props;
import akka.actor.SupervisorStrategy;
import akka.actor.SupervisorStrategy.Directive;
import akka.actor.UntypedActor;
import akka.japi.Function;

public class SupervisorActor extends UntypedActor {

	private final  ActorRef childActor;

	public SupervisorActor() {
		childActor = getContext().actorOf(Props.create(WorkerActor.class),
				"workerActor");
	}

	private static SupervisorStrategy strategy = new OneForOneStrategy(10,
			Duration.create("10 second"), new Function<Throwable, Directive>() {
				public Directive apply(Throwable t) {
					if (t instanceof ArithmeticException) {
						return resume();
					} else if (t instanceof NullPointerException) {
						return restart();
					}  else {
						return escalate();
					}
				}
			});

	@Override
	public SupervisorStrategy supervisorStrategy() {
		return strategy;
	}

	public void onReceive(Object o) throws Exception {
		if (o instanceof ResultMessageRequest) {
			childActor.tell(o, getSender());
		} else
			childActor.tell(o,ActorRef.noSender());
	}

	public ActorRef getChildActor() {
		return childActor;
	}

}

package com.poc.acteur.supervisor;

import java.util.concurrent.TimeUnit;

import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.pattern.Patterns;

/**
 * Programme principal
 * 
 */
public class Main {

	public static void main(String[] args) throws Exception {
		
		ActorSystem system = ActorSystem.create("faultTolerance");
		LoggingAdapter log = Logging.getLogger(system, system);
		
		//creation de l'acteur superviseur
		ActorRef supervisor = system.actorOf(
				Props.create(SupervisorActor.class), "supervisor");

		ResultMessageRequest request = new ResultMessageRequest();
		
		//Envoi d'un entier au superviseur
		log.info("Envoi de  l'entier 2 ...");
		supervisor.tell(new IntegerMessage (2),ActorRef.noSender());
		
		//Future pour recuperer  reponse provenant du superviseur
		Future<Object> future = Patterns
				.ask(supervisor, request, 5000);
		Integer result = (Integer) Await.result(
				future, Duration.create(5000, TimeUnit.MILLISECONDS));

		//La valeur attendu est 4/2 -> 2
		log.info("Valeur recue-> {}", result);
		assert result.equals(Integer.valueOf(2));
		
		log.info("Envoi de  l'entier 0 ...");
		supervisor.tell(new IntegerMessage (0),ActorRef.noSender());
		
		//Future pour recuperer  reponse provenant du superviseur
		 future = Patterns
				.ask(supervisor, request, 5000);
		 result = (Integer) Await.result(
				future, Duration.create(5000, TimeUnit.MILLISECONDS));

		log.info("Valeur recue-> {}", result);
		
		//L'exception ArithmetiqueException a ete declenchee par le worker
		//Le worker continue donc la valeur renvoyee est toujours 2
		assert result.equals(Integer.valueOf(2));
		
		log.info("Envoi de  l'entier wrappe avec une valeur null ...");
		supervisor.tell(new IntegerMessage (null),ActorRef.noSender());
		
		//Future pour recuperer  reponse provenant du superviseur
		 future = Patterns
				.ask(supervisor, request, 5000);
		 result = (Integer) Await.result(
				future, Duration.create(5000, TimeUnit.MILLISECONDS));

		//L'exception NullPointerException a ete declenchee par le worker
		//Le worker est redemaree la valeur renvoyee est la valeur initiale 4
		log.info("Valeur recue-> {}", result);
		assert result.equals(Integer.valueOf(4));

		log.info("Worker Actor shutdown !");
		system.shutdown();

	}
}