Flink之间的组件通信

Posted 2023-02-23 张少凯

Flink内部节点之间的通信是用Akka,比如jobmanager和taskmanager之间的通信。而operator之间的数据传输是用netty。一句话总结，组件之间的传递用的akka，数据之间的网络传输用的是netty。

  flink通过akka进行的分布式通信的实现，在0.9版本使用。使用akka，所有远程过程调用现在都实现为异步消息。
  RPC框架是flink任务运行的基础，flink整个RPC框架都是基于akka实现，并对akka中的actorsystem、actor进行了封装和使用。

RPC,netty，akka三者之间的关系？

  RPC是一个统称，全称叫做Remote Procedure Call，远程方法调用。是一个广泛的概念。是akka或者netty具体的某一种实现。

akka与actor模型

akka是一个并发、容错和可伸缩应用的框架。基本所有框架都有这几个优点。
每个actor都是一个单一的线程，它不断地从其邮箱中poll（拉取）消息，并且连续不断地处理，可以改变它自身的内部状态，

actor系统

可以理解为一个工作部门，每一个actor是一个员工，整个是一个actor系统。一个actor系统包含了所有存活的actors。
多个actor系统可以在一个机器上共存。actor系统能够自动识别消息是发送给本地机器还是远程机器的actor系统。
所有actors都是通过继承的方式来组织的。每个新创建的actor为子actor，每个父actor对自己的子actor负责监督。如果子actor出现错误，父actor将会收到通知。

flink中的actors

actor是一个包含状态和行为的容器。
flink系统由三个分布式组件构成：jobclient，jobmanager，taskmanager。
jobclient从用户处得到flink job，提交给jobmanager。jobmanager策划这个job的执行，首先分配所需的资源，分配的资源主要就是taskmanager上要执行的slot。资源分配之后，jobmanager部署单独的任务到响应的task manager上，一旦收到一个任务，task manager产生一个线程用来执行这个任务。状态的改变（如开始计算或者完成计算）将被发送回jobmanager。基于这些状态的更新，jobmanager将引导这个job的执行直到完成。一旦一个job执行完成，其结果就会被发送回jobclient。

异步VS消息同步

在任何地方，flink使用异步消息和通过futures（用来获取异步的响应）来处理响应。futures有一个超时时间，以防止操作失败。这是为了防止死锁，超市时间可以通过“akka.timeout”来配置。
两个actor在通信之前，需要获取一个actorRef，通过这个引用来和另一个actor通信。这个操作的查找也有一个超时时间，防止查询超时，通过"akka.lookup.timeout"。
akka可以限制消息的大小，可通过"akka.framesize"配置来更改大小。

使用akka

akka系统的核心是actorsystem和actor，actor不能直接创建，必须先创建actorsystem才能创建actor，另外，我们只能通过actorRef跟actor进行通信。
akka有两种核心的异步通信方式：tell和task。tell：仅仅使用异步方式给某个actor发送消息，无需等待actor响应结果，并且也不会阻塞后面代码的运行。
ask：当我们需要从actor获取响应结果时，可使用ask方法，ask方法会将返回结果包装到类scala.concurrent.Future中。然后通过异步回调获取返回结果。
上面主要介绍了akka中的actorsystem、actor，以及与actor的通信，flink借此构建其底层通信系统。

RPCGateway网关

flink的RPC协议通过RPCgateway来定义，主要定义通信行为，用于远程调用RPCendpoint方法，可以理解为对方的代理。job manager---》gateway（包含了各种行为方法，控制task manager）---》task manager。

RPCEndpoint终端

RpcEndpoint是通信终端，提供RPC服务组件的生命周期管理（start、stop）。每个RPCendpoint对应了一个路径（endpoint和actorsystem共同确定），每个路径对应一个actor，它实现了RpcGateway接口。
构造的时候调用rpcService.startServer()启动RpcServer，进入可以接收请求的状态，最后将RpcServer绑定到主线程上真正执行起来。
在RpcEndpoint中还定义了一些方法如runAsync(Runnable)、callAsync（callable，time）方法来执行Rpc调用，值得注意的是在flink的设计中，对于同一个endpoint，所有的调用都运行在主线程，因此不会有并发问题，当启动Rpcendpoint进行RPC通信时，它会委托RpcServer进行处理。

RpcService和RpcServer

RpcService和RpcServer是RPCendpoint的成员变量。
RpcService是Rpc服务的接口，其作用如下：
                                    根据提供的RPCendpoint来启动和停止RpcServer（actor）
                                    根据提供的地址链接到对方的RpcServer，并返回一个Rpcgateway
                                    延迟\\立刻调度runnable、callable
在flink中的实现类为AkkaRpcService，是Akka的ActorSystem的封装，基本可以理解为ActorSystem的一个适配器。
最终使用动态代理将所有的消息转发到InvocationHandler。
RpcServer负责接收响应远端RPC消息请求，是一个自身的代理对象（终端的启动实际上是由自身网关RpcServer来启动的rpcserver.start()方法。）。有两个实现：AkkaInvocationHandler和FencedAkkaInvocationHandler。

AkkaRpcActor

AkkaRpcActor是Akka的具体实现，主要负责处理如下类型消息：
      1.本地Rpc消息，调用LocalRpcInvocation
            会指派给rpcEndpoint进行处理，如果有响应，则将结果返还给sender。
      2.runAsync消息 && callAsync消息
            这类消息带有可执行的代码，直接在actor的线程中执行。
      3.控制消息controlMessages
            用来控制actor行为，start启动，stop停止，停止后收到的消息会丢弃。

RPC交互过程

RPC通信过程分为请求和响应。
请求：在RPCservice中调用connect（）方法与对端的RPCendpoint（RPCserver）建立连接，connect（）方法根据返回的地址返回InvocationHandler（AkkaInvocationHandler或FencedAkkaInvocationHandler）。
响应：RPC消息通过RPCendpoint所绑定的actor的actorREF发送的，akkaRPCactor是消息的接收入口，akkaRPCactor在RPCendpoint中构造生成，负责将消息交给不同的方法进行处理。

RPC流程图