Actor模型与Akka

一. Actor模型

Actor模型概念

一个概念模型,用于处理并发计算
Actor模型内部的状态由自己的行为维护,外部线程不能直接调用对象的行为,必须通过消息才能激发行为,这样就保证Actor内部数据只有被自己修改

Actor模型组成

Actor模型详解

在Actor模型中,一切都可以抽象为Actor,一个Actor指的是一个最基本的计算单元,它能接收一个消息并且基于该消息执行计算
Actor封装状态和行为的对象,他们的唯一通讯方式就是交换消息,类比Object: 一个对象接收一条消息(方法调用),然后根据接收的消息做事(调用了哪个方法)
Actor重要特征在于Actors之间相互隔离,他们并不互相共享内存,这点儿又区别于上面的Object对象,也就是说一个Actor能维持一个私有的状态,并且这个状态不可能被另一个Actor改变
Actor交换的消息放在接收方的邮箱(Inbox)里,也就是说Actor之间并不直接通信,而是通过了消息来相互沟通,每一个Actor都把它要做的事情都封装在了它的内部
每一个Actor是可以有状态也可以是无状态的
当一个Actor接收到消息后,能做如下三件事中的一件: 创建其它Actors,向其它Actors发送消息,指定下一条消息到来的行为:就是Actors如何修改状态
每一个Actor都拥有属于自己的轻量级线程,保护它们不会被系统中的其他部分影响,因此,我们在编写Actor时,就不用担心并发的问题
通过Actor能够简化锁以及线程管理:
- 提供一种高级的抽象,能够封装状态和操作,简化开发应用开发
- 提供了异步的非阻塞的/高性能的事件驱动模型
- 超级轻量级的线程事件处理能力

Actor系统

Actor的生命周期

ActorRef
它是Actor的一个引用,作用是发送消息给它表示的Actor, 而Actor可以通过self()或sender()方法来获取到自身或者消息发送者的Actor引用,在Actor系统中,Actor之间永远都不能直接的通信,必须通过他们的代理ActorRef建立通信

Actor路径

父子Actor
树形结构,分层管理,通过递归把复杂的事务原子化
Actor Path:Actor System中唯一标识
- 本地Actor
  "akka://my-sys/user/service-a/worker1"
- 远程Actor
  "akka.tcp://my-sys@host.example.com:2552/user/service-b"
- 集群Actor
  "cluster://my-cluster/service-c"
Actor Path的默认节点
- root guardian是所有Actor的父Actor /
- UserActor是所有用户创建的Actor的父Actor /user
- SystemActor是所有系统创建的Actor的父Actor /system
- 同级Actor /*

二. Akka

Akka是一个由Scala编写的,兼容Scala和Java,用于编写高可用和高伸缩性的Actor模型框架
它基于了事件驱动的并发处理模式,性能非常高,并且有很高的可用性
简化应用系统中开发并发处理的过程

Akka的开发过程

定义消息模型
创建Actor
- 实现业务逻辑
获取Actor
- 需要使用Akka的地方获取到ActorSystem,然后根据业务的数据流,获取到合适的Actor,给Actor发送消息
消息处理
- 在Actor的实现中对接收到的消息进行具体的处理或转发,从而形成业务逻辑流

创建Actor

TypedActor
- Active Object设计模式的应用: 当调用这个实例的方法,异步执行而不是同步执行
- TypedActor并不使用消息,更多的是用来桥接Actor系统和非Actor
UnTypedActor
- 更像是JAVA中的JMS调用
- 方法的调用和执行完全依赖消息,通过消息的类型或内容来区别不同的执行
在Akka中我们更多的倾向于使用UnTypedActor向Actor系统间传递消息

创建UnTypedActor

    public class GreetPrinter extends UntypedActor{@overridepublic void onReceive(Object message) throws Exception{if(message instanceof Greeting)System.out.println(((Greeting) message).message);}}

获取Actor

三种方法:

actorOf
创建一个新的Actor,创建的Actor为调用该方法时所属的Context下的直接子Actor
该方法会返回一个ActorRef来表示Actor的引用,包含一个UID和一个Path.这两个值共同的标识了一个Actor的唯一性,ActorRef的生命周期在Actor停止的时候结束
actorSelection
当消息传递来时,只查找现有的Actor,而不会创建新的Actor,也不会验证目标Actors是否存在
该方法值关心Path而不关心具体是哪一个Actor,如果要通过ActoSelection来获取一个具体的Actor,需要调用ActorSelection的resolveOne的方法来获取
actorFor
只会查找现有的Actor,而不会创建新的Actor

Akka其它组件

Dispatcher 分发器

当Actor的数量比较多后,彼此之间的通信就需要协调,从而能更好的平衡整个系统的执行性能
特征: 在自己独立的线程上不断的进行协调,把来自各个Actor的消息分配到执行线程
四种不同的Dispatcher

Dispatcher:可以被多个Actor所共享,通过ThreadPool实现

    private List<ActorRef> createActors(int actorCount) { Props props = Props.create(WriterActor.class).withDispatcher("writer-dispatcher);List<ActorRef> actors = new ArrayList<>(actorCount);for(int i=0;i<actorCount;i++){actors.add(getContext().actorOf(props,"writer_"+i));}return actors;}

PinnedDispatcher
BalancingDispatcher
CallingThreadDispatcher

Router 路由器

当有多个目标时,Akka通过Router机制,来有效的分配消息给Actor来完成工作,被Router管理的Actor被称作Routee

Router的使用:

可以在Actor内通过实例化Router对象的方式来使用
也可以在Actor外通过withRouter的方式直接创建一个RouterActor来使用

Router router = new Router(new RoundRobinRoutingLogic());for(ActorRef actor : actors){
router = router.addRoutee(actor);
}
router.route("Insert",ActorRef.noSender());

Router路由策略
◦ RoundRobinPool:轮询方式分发消息
◦ RandomPool:随机方式分发消息
◦ BalancingPool:均衡分发消息
◦ SmallestMailboxPool:最少消息邮箱分发
◦ BroadcastPool:广播消息
◦ ScatterGatherFirstCompletedPool:所有子Routee发送,第一个消息回复
◦ TailChoppingPool:随机延迟发送,第一个回复消息
◦ ConsistentHashingPool:一致性hash的方式来分发消息

Scheduler 调度器

Akka Demo

https://github.com/sunxiang0918/AkkaDemo

三. Akka分布式消息及在Spark中的应用

Akka远程调用

采用了端对端(peer-to-peer, P2P)的通信方式设计,简单透明:除了传递的消息需要可序列化以及创建和查找Actor的时候路径稍有不同外,没有其他的区别,Akka没有特别为Remoting提供专门的API,区别在于配置
最小配置:

    akka{actor{provider = "akka.remote.RemoteActorRefProvider"}remote{enabled-transports = ["akka.remote.netty.tcp"]netty.tcp{hostname = "127.0.0.1"port = 2552}}}

远程Actor

创建和获取远程Actor

远程路径:
akka.tcp://WCMapReduceApp@127.0.0.1:2552/user/remoteActor
创建方式:

    ActorSystem system = ActorSystem.create("WCMapReduceApp",ConfigFactory.load("application").getConfig("WCMapReduceApp"));aggregateActor = system.actorOf(Props.create(AggregateActor.class));

获取方式:

    final ActorRef remoteActor = system.actorFor("akka.tcp://WCMapReduceApp@127.0.0.1:2552/user/remoteActor")

远程Actor路由

负载均衡和高可用
Akka远程调用是基于点对点模型,不支持负载均衡等功能,但能通过Actor路由机制实现负载均衡和高可用
ex: 当向/parent/remotePool发送消息的时候,,会轮询的把消息分发到不同的远程服务上

    akka.actor.deployment {/parent/remotePool{router = round-robin-poolnr-of-instances = 10target.nodes = ["akka.tcp://app@10.0.0.2:2552","akka://app@10.0.0.3:2552"]}}

Akka集群

Cluster调用
Gossip协议
反熵(Anti-Entropy): 在一个有界网络中,每个节点都随机地与其他节点通信,经过一番杂乱无章的通信,最终所有节点的状态都会达成一致
,具有去中心化,节点之间完全对等,不需要任何的中心节点,最终一致性的特点
Gossip算法示例
- Cassandra
- redis

Akka的集群是基于Gossip协议实现的,支持服务自动失效检测,能够自动发现出现问题而离开集群的成员节点,通过事件驱动的方式,将状态传播到整个集群的其他成员节点中去,一个Akka集群是由一组成员节点组成的,每个成员节点都是通过hostname:port:uid来唯一标识,并且每个成员节点完全解耦合

节点状态
集群Leader
集群配置
Spark中的Akka
把Akka当做分布式系统中的RPC框架使用,很多组件封装为Actor,进行控制和状态通信,Spark中的CLient,Master和Worker都是一个Actor,Actor之间,消息发送端通过”!”符号发送消息,接收端通过receive方法中的case模式匹配接收和处理消息