沧州市科一网站,常宁市网站建设,建设网银登录官网,电商网站开发技术文章目录 1. Akka基本概念与Actor模型2. Akka相关demo2.1. 创建Akka系统2.2. 根据path获取Actor并与之通讯 3. Flink RPC框架与Akka的关系4.运行时RPC整体架构设计5. RpcEndpoint的设计与实现 我们从整体的角度看一下Flink RPC通信框架的设计与实现#xff0c;了解其底层Akka通… 文章目录 1. Akka基本概念与Actor模型2. Akka相关demo2.1. 创建Akka系统2.2. 根据path获取Actor并与之通讯 3. Flink RPC框架与Akka的关系4.运行时RPC整体架构设计5. RpcEndpoint的设计与实现 我们从整体的角度看一下Flink RPC通信框架的设计与实现了解其底层Akka通信框架的基础概念及二者之间的关系。 1. Akka基本概念与Actor模型
Akka是使用Scala语言编写的库用于在JVM上简化编写具有可容错、高可伸缩性的Java或Scala的Actor模型。Akka基于Actor模型提供了一个用于构建可扩展、弹性、快速响应的应用程序的平台。
Actor 模型是一种并发计算模型Actor 模型的核心思想是将计算单元抽象为独立的并发实体称为 “actors”这些 actors 之间通过消息传递进行通信。
以下是 Actor 模型的一些关键概念 ActorActor 是计算模型的基本执行单元。每个 Actor 都有自己的状态、行为和邮箱用于接收消息。Actor 之间是相互独立的它们通过消息传递进行通信。消息传递在 Actor 模型中通信是通过消息传递来实现的。一个 Actor 可以向另一个 Actor 发送消息消息包含了要执行的操作或者改变状态的请求。这种异步消息传递使得系统更具有弹性和可伸缩性。地址每个 Actor 都有一个唯一的地址用于唯一标识该 Actor。其他 Actor 可以使用地址向目标 Actor 发送消息。邮箱每个 Actor 都有一个邮箱用于存储接收到的消息。Actor 处理消息的速度可能不同但由于消息传递是异步的这不会阻塞发送者。行为Actor 的行为定义了对消息的响应方式包括状态的修改、消息的处理等。行为可以随着时间和接收到的消息而动态变化。 Actor由状态State、行为Behavior和邮箱Mailbox三部分组成。 actors和其他actors通过发送异步消息通信。Actor模型的强大来自于异步。它也可以显式等待响应这使得可以执行同步操作。但是强烈不建议同步消息因为它们限制了系统的伸缩性怎么实现的伸缩性。 actor系统 每个actor是一个单一的线程它不断地从其邮箱中poll(拉取)消息并且连续不断地处理。对于已经处理过的消息的结果actor可以改变它自身的内部状态或者发送一个新消息或者孵化一个新的actor。 2. Akka相关demo
2.1. 创建Akka系统
Akka系统的核心组件包括ActorSystem和Actor构建一个Akka系统首先需要创建ActorSystem然后通过ActorSystem创建Actor。 需要注意的是 Akka不允许直接创建Actor实例只能通过ActorSystem.actorOf和ActorContext.actorOf等特定接口创建Actor。只能通过ActorRef与Actor进行通信ActorRef对原生Actor实例做了良好的封装外界不能随意修改其内部状态。 如代码所示Akka系统中包含了创建ActorSystem以及Actor的基本实例。
// 1. 构建ActorSystem
// 使用缺省配置
ActorSystem system ActorSystem.create(sys);
// 也可显示指定appsys配置
// ActorSystem system1 ActorSystem.create(helloakka, ConfigFactory.load(appsys));// 2. 构建Actor,获取该Actor的引用即ActorRef
ActorRef helloActor system.actorOf(Props.create(HelloActor.class), helloActor);// 3. 给helloActor发送消息
helloActor.tell(hello helloActor, ActorRef.noSender());// 4. 关闭ActorSystem
system.terminate();
在Akka中创建的每个Actor都有自己的路径该路径遵循 ActorSystem 的层级结构大致如下
本地akka://sys/user/helloActor
远程akka.tcp://sysl27.0.0.1:2020/user/remoteActor - sys创建的ActorSystem的名字
- user通过ActorSystem#actorOf和ActorContext#actorOf 方法创建的 Actor 都属于/user下其是系统层面创建的与系统整体行为有关在开发阶段并不需要对其过多关注
- helloActor我们创建的HelloActor。其中远程部分路径含义如下- akka.tcp远程通信方式为tcp
- sys127.0.0.1:2020ActorSystem名字及远程主机ip和端口号。 2.2. 根据path获取Actor并与之通讯
若提供了Actor的路径可以通过路径获取到ActorRef然后与之通信代码如下所示
ActorSystem system ActorSystem.create(sys)
ActorSelection as system.actorSelection(/path/to/actor);Timeout timeout new Timeout(Duration.create(2, seconds));
FutureActorRef fu as.resolveOne(timeout);fu.onSuccess(new OnSuccessActorRef() {Overridepublic void onSuccess(ActorRef actor) {System.out.println(actor: actor);actor.tell(hello actor, ActorRef.noSender());}
}, system.dispatcher());fu.onFailure(new OnFailure() {Overridepublic void onFailure(Throwable failure) {System.out.println(failure: failure);}
}, system.dispatcher()); 3. Flink RPC框架与Akka的关系
Flink进行RPC通信的组件 如图所示从Flink RPC节点关系中可以看出集群运行时中实现了RPC通信节点功能的主要有Dispatcher、ResourceManager和TaskManager以及JobMaster等组件。 借助RPC通信这些组件共同参与任务提交及运行的整个流程例如通过客户端向Dispatcher服务提交JobGraphJobManager向TaskManager提交Task请求以及TaskManager向JobManager更新Task执行状态等。 通过AkkaRpcService实现远程通讯能力 从图中也可以看出集群的RPC服务组件是1RpcEndpoint每个RpcEndpoint包含一个内置的RpcServer负责执行本地和远程的代码请求2RpcServer对应Akka中的Actor实例。RpcEndpoint中创建和启动RpcServer主要是基于集群中的3RpcService实现4RpcService的主要实现是AkkaRpcService。 从图可以看出AkkaRpcService将Akka中的ActorSystem进行封装通过AkkaRpcService可以创建RpcEndpoint中的RpcServer同时基于AkkaRpcService提供的connect()方法与远程RpcServer建立RPC连接提供远程进程调用的能力。 4.运行时RPC整体架构设计
Flink的RPC框架设计非常复杂除了基于Akka构建了底层通信系统之外还会使用JDK动态代理构建RpcGateway接口的代理类。 Flink RPC UML关系图
这里我们简单梳理一下RPC架构涉及的组件以及每种组件的作用。 集群RPC组件的基本实现类 RpcEndpoint提供了集群RPC组件的基本实现所有需要实现RPC服务的组件都会继承RpcEndpoint抽象类。 RpcEndpoint中包含了endpointId用于唯一标记当前的RPC节点。RpcEndpoint借助RpcService启动内部RpcServer之后通过RpcServer完成本地和远程线程执行。 基本实现类与FencedToken对比 对于RpcEndpoint来讲底层主要有FencedRpcEndpoint基本实现类。 实现FencedRpcEndpoint的RPC节点都会有自己的FencedToken当进行远程RPC调用时会对比访问者分配的FencedToken和被访问者的FencedToken结果一致才会进行后续操作。 RpcEndpoint的实现类有TaskExecutor组件FencedRpcEndpoint的实现类有Dispatcher、JobMaster以及ResourceManager等组件。这些组件可以获取RpcService中ActorSystem的dispatcher服务并直接通过dispatcher创建Task线程实例。RpcService提供了创建和启动RpcServer的方法。 在启动RpcServer的过程中通过RpcEndpoint的地址创建Akka Actor实例并基于Actor实例构建RpcServer接口的动态代理类向RpcServer的主线程中提交Runnable以及Callable线程等。 同时在RpcService中提供了连接远程RpcEndpoint的方法并创建了相应RpcGateway接口的动态代理类用于执行远程RPC请求。 RpcServer接口通过AkkaInvocationHandler动态代理类实现所有远程或本地的执行请求最终都会转换到AkkaInvocationHandler代理类中执行。 AkkaInvocationHandler实现了MainThreadExecutable接口提供了runAsync(Runnable runnable)以及callAsync(CallableV callable, Time callTimeout)等在主线程中执行代码块的功能。例如在TaskExecutor中释放Slot资源时会调用runAsync()方法将freeSlotInternal()方法提交到TaskExecutor对应的RpcServer中运行此时就会调用AkkaInvocationHandler在主线程中执行任务. 5. RpcEndpoint的设计与实现
RpcEndpoint是集群中RPC组件的端点每个RpcEndpoint都对应一个由endpointId和actorSystem确定的路径且该路径对应同一个Akka Actor。
如图所有需要实现RPC通信的集群组件都会继承RpcEndpoint抽象类例如TaskExecutor、Dispatcher以及ResourceManager组件服务还包括根据JobGraph动态创建和启动的JobMaster服务。
从图中我们可以看出RpcEndpoint实现了RpcGateway和AutoCloseableAsync两个接口其中 RpcGateway 提供了动态获取RpcEndpoint中Akka地址和HostName的方法。 因为JobMaster组件在任务启动时才会获取Akka中ActorSystem分配的地址信息所以借助RpcGateway接口提供的方法就能获取Akka相关连接信息。 RpcEndpoint中包含RpcService、RpcServer以及MainThreadExecutor三个重要的成员变量其中 RpcService是RpcEndpoint的后台管理服务RpcServer是RpcEndpoint的内部服务类MainThreadExecutor封装了MainThreadExecutable接口其主要底层实现是AkkaInvocationHandler代理类。所有本地和远程的RpcGateway执行请求都会通过动态代理的形式转换到AkkaInvocationHandler代理类中执行。
