网站建设行吗/seo数据优化

什么是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);}@Overridepublic SupervisorStrategy supervisorStrategy() {//TODO 这里覆盖父类的 supervisorStrategy，可以指定自己的监督策略//这里为了示意，仅仅返回一个null，有两种策略，有AllForOneStrategy和OneForOneStrategyreturn null;};@Overridepublic 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被重启后执行的@Overridepublic void preStart(){System.out.println("I am going to work, in preStart method.");}//Actor停止时，postStop()方法会被调用，同时这个Actor的监视者会收到//一个Terminated消息。@Overridepublic void postStop(){System.out.println("MyWorker is stoping.");}@Overridepublic 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 发送到 workerWork 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方法就是消息处理器@Overridepublic 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 listenerDuration duration = Duration.create(System.currentTimeMillis() - start, TimeUnit.MILLISECONDS);listener.tell(new PiApproximation(pi, duration), getSelf());// Stops this actor and all its supervised childrengetContext().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 systemfinal ActorRef listener = system.actorOf(new Props(Listener.class), "listener");// create the masterActorRef 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());}
}