akka introduce
高弘光
什么是Akka，Akka主要用来编写分布式的，可扩展的，并发的应用程序，详见官方文档，官方文档真的好长啊，这里记了一些akka的helloworld程序，算是简单入门吧。
Akka总结
Actor模型基于一种通过消息进行通信的简单的actor实体结构，能够轻松地构建应用程序来实现高并发性和可伸缩性。
用于并发计算的actor模型基于各种称为actor的原语来构建系统。Actor执行操作来响应称为消息的输入。这些操作包括更改actor自己的状态，以及发出其他消息和创建其他actor。所有消息都是异步交付的，因此将消息发送发与接收方分开。正是由于这种分离，导致actor系统具有内在的并发性：可以不受限制地并行执行任何拥有输入消息的actor。在Akka中的消息都是不可变对象（这个在之前已经了解过了，包括怎么创建一个不可变对象）。
Akka的监督策略有OneForOneStrategy和AllForOneStrategy，对于OneForOneStrategy策略，子Actor只会对出问题的子Actor进行处理，比如重启或者停止，而对于AllForOneStrategy，父Actor会对出问题的子Actor以及它所有的兄弟都进行处理。很显然，对于AllForOneStratege，它更适合各个Actor联系非常紧密的场景，如果多个Actor间只有一个Actor出现故障，则宣告整个任务失败。
选择Actor，主要是通过通配符（actor的名字）来进行Actor选择。
第一个例子，来自《实战Java高并发...》，重点是一个Actor WatchDog监督另一个Actor MyWorker，在被监督的Actor退出时，监督者会收到一个Terminated类型的消息，另外Actor也可以重写supervisorStrategy() 方法来指定监督策略。其他一些细节都在程序中说明了，代码如下：
Main类：
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
public class DeadMain {
public static void main(String[] args) {
//ActorSystem.create()的第一个参数为系统名称，第一个参数为配置文件，这里没有配置文件
ActorSystem system = ActorSystem.create("deadwatch");
//ActorSystem.actorOf是创建一个Actor的方式，第二个参数为 actor的名字
final ActorRef worker = system.actorOf(new Props(MyWorker.class));
//当创建的actor的构造方法有参数时，可以使用下的方式创建
ActorRef watcher = system.actorOf(new Props(new UntypedActorFactory() {
private static final long serialVersionUID = 1L;
public UntypedActor create() {
return new WatchActor(worker);
}
}),"awatcher");
//前两种方式向worker发送消息，没有发送者，后一种方式的发送者为watcher。
worker.tell(MyWorker.Msg.WORKING);
worker.tell(MyWorker.Msg.DONE);
worker.tell(MyWorker.Msg.CLOSE, watcher);
}
}
一个监督Actor WatchActor：
import akka.actor.ActorRef;
import akka.actor.SupervisorStrategy;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
//无类型的Actor UntpyedActor，继承了这个就不能继承其他类了，这个是不足
//还有一种有类型的Actor
public class WatchActor extends UntypedActor{
//构造函数
public WatchActor(ActorRef ref){
getContext().watch(ref);
}
@Override
public SupervisorStrategy supervisorStrategy() {
//TODO 这里覆盖父类的 supervisorStrategy，可以指定自己的监督策略
//这里为了示意，仅仅返回一个null，有两种策略，有AllForOneStrategy和OneForOneStrategy
return null;
};
@Override
public void onReceive(Object msg) throws Exception {
//如果被监视的对象（即MyWorker）退出，WatchActor将会收到一条Terminated消息。
//这个消息不是MyWorker显示发送的，系统发送的？
if(msg instanceof Terminated){
System.out.println(
String.format("%s has terminated, shutdown system.", ((Terminated)msg).getActor().path()));
getContext().system().shutdown();
}
else if(msg==MyWorker.Msg.CLOSE){ //这个消息是在MyWorker退出的时候显示发送的
System.out.println("I receive a message from sub actor, he said close");
}else{
unhandled(msg);
}
}
}
一个工作actor：
import akka.actor.UntypedActor;
//一个重写了各个生命周期的actor
public class MyWorker extends UntypedActor{
public static enum Msg{
WORKING,DONE,CLOSE;
}
//actor创建以后，会调用preStart()方法，在这个方法里，会进行一些资源的初始化工作
//另外还有一个preRestart和postRestart方法，是在Actor被重启后执行的
@Override
public void preStart(){
System.out.println("I am going to work, in preStart method.");
}
//Actor停止时，postStop()方法会被调用，同时这个Actor的监视者会收到
//一个Terminated消息。
@Override
public void postStop(){
System.out.println("MyWorker is stoping.");
}
@Override
public void onReceive(Object msg) throws Exception {
if(msg==Msg.WORKING){
//这里打印getSender()输出是 deadLetters
//没有发送者 就是 deadLetters ?
System.out.println(getSender());
System.out.println("I am working");
}
if(msg==Msg.DONE){
System.out.println("stop working");
}
if(msg==Msg.CLOSE){
System.out.println("I will shutdown");
//这里打印getSender()输出是awatcher，就是watcher actor的名字
System.out.println(getSender());
//向这个消息（Msg.CLOSE）的发送者再发一条消息，告诉自己即将死了
getSender().tell(Msg.CLOSE,getSelf());
//这里调用context().stop()方法来停止一个actor，停止一个
//actor的方式很多。
context().stop(getSelf());
}else{
unhandled(msg);
}
}
}下面是另一个例子，来自官方文档，计算圆周率Pi的，几个worker actor并行工作，感觉也很经典。
import java.util.concurrent.TimeUnit;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
import akka.routing.RoundRobinRouter;
import akka.util.Duration;
public class Pi {
public static void main(String[] args) {
Pi pi = new Pi();
pi.calculate(4, 1000000, 10000);
}
//下面这四个静态内部类是4中不同类型的消息
//Calculate，发送到Master，以启动计算
static class Calculate {}
//从Master actor发送到 worker actor，并携带任务
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;
}
}
//从worker actor发送到 Master actor并携带结果
static class Result {
private final double value;
public Result(double value) {
this.value = value;
}
public double getValue() {
return value;
}
}
//从master actor发送到 Listener actor并携带最终结果
static class PiApproximation {
private final double pi;
private final Duration duration;
public PiApproximation(double pi, Duration duration) {
this.pi = pi;
this.duration = duration;
}
public double getPi() {
return pi;
}
public Duration getDuration() {
return duration;
}
}
//工作节点，只从mater actor接受一种类型的参数
public static class Worker extends UntypedActor {
//执行具体的计算任务，在worker中定义，此方法被封起来了
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;
}
public void onReceive(Object message) {
if (message instanceof Work) { //Work消息，从master 发送到 worker
Work work = (Work) message;
double result = calculatePiFor(work.getStart(), work.getNrOfElements());
getSender().tell(new Result(result), getSelf());
} else {
//这个方法是在父类中定义的
unhandled(message);
}
}
}
//控制节点，接受两种类型的参数，从Listener发送的，从worker节点发送的
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;
//这里创建了一个actor，并为其指定了一个路由策略(轮询)
workerRouter = this.getContext()
.actorOf(new Props(Worker.class).withRouter(new RoundRobinRouter(nrOfWorkers)), "workerRouter");
}
//onReceive方法就是消息处理器
@Override
public void onReceive(Object message) {
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);
}
}
}
//接收从master发来的最终结果，并打印输出
//这个Actor的构造不需要构造参数
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);
}
}
}
//这个方法是整个计算开始的方法，第一个参数表示worker进程的个数
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), "listener");
// create the master
ActorRef master = system.actorOf(new Props(new UntypedActorFactory() {
private static final long serialVersionUID = 1L;
public UntypedActor create() {
return new Master(nrOfWorkers, nrOfMessages, nrOfElements, listener);
}
}), "master");
// start the calculation
//谁调用tell，就是向谁发送？？
master.tell(new Calculate());
}
}
类似资料：