Flink内部节点之间的通信是用Akka，比如JobManager和TaskManager之间的通信。而operator之间的数据传输是利用Netty。

Flink通过Akka进行的分布式通信的实现，在0.9版中采用。使用Akka，所有远程过程调用现在都实现为异步消息。这主要影响组件JobManager，TaskManager 和JobClient。将来，甚至有可能将更多的组件转换为参与者，从而允许它们发送和处理异步消息。

RPC框架是Flink任务运行的基础，Flink整个RPC框架基于Akka实现，并对Akka中的ActorSystem、Actor进行了封装和使用，Flink整个通信框架的组件主要由RpcEndpoint、RpcService、RpcServer、AkkaInvocationHandler、AkkaRpcActor等构成。RpcEndpoint定义了一个Actor的路径；RpcService提供了启动RpcServer、执行代码体等方法；
RpcServer/AkkaInvocationHandler提供了与Actor通信的接口；AkkaRpcActor为Flink封装的Actor。

Akka与Actor模型

Akka是一个开发并发、容错和可伸缩应用的框架。它是Actor Model的一个实现，和Erlang的并发模型很像。在Actor模型中，所有的实体被认为是独立的actors。actors和其他actors通过发送异步消息通信。Actor模型的强大来自于异步。它也可以显式等待响应，这使得可以执行同步操作。但是，强烈不建议同步消息，因为它们限制了系统的伸缩性。每个actor有一个邮箱(mailbox)，它收到的消息存储在里面。另外，每一个actor维护自身单独的状态。一个Actors网络如下所示：

每个actor是一个单一的线程，它不断地从其邮箱中poll(拉取)消息，并且连续不断地处理。对于已经处理过的消息的结果，actor可以改变它自身的内部状态或者发送一个新消息或者孵化一个新的actor。尽管单个的actor是自然有序的，但一个包含若干个actor的系统却是高度并发的并且极具扩展性的。因为那些处理线程是所有actor之间共享的。这也是我们为什么不该在actor线程里调用可能导致阻塞的“调用”。因为这样的调用可能会阻塞该线程使得他们无法替其他actor处理消息。

1、 Actor系统

一个Actor系统包含了所有存活的actors。它提供的共享服务包括调度、配置和日志等。Actor系统同时包含一个线程池，所有actor从这里获取线程。

多个Actor系统可以在一台机器上共存。如果一个Actor系统通过RemoteActorRefProvider启动，它就可以被其他机器上的Actor系统发现。Actor系统能够自动识别消息是发送给本地机器还是远程机器的Actor系统。在本地通信的情况下，消息通过共享存储器高效地传输。在远程通信的情况下，消息通过网络栈发送。

所有Actors都是继承来组织的。每个新创建的actor将其创建的actor视作父actor。继承被用来监督。每个父actor对自己的子actor负责监督。如果在一个子actor发生错误，父actor将会收到通知。如果这个父actor可以解决这个问题，它就重新启动这个子actor。如果这个错误父actor无法处理，它可以把这个错误传递给自己的父actor。

第一个actor通过系统创建，由/user 这个actor负责监督。详细的Actor的继承制度可以参考
https://doc.akka.io//docs/akka/snapshot/general/supervision.html。

2、 Flink中的Actor

Actor是一个包含状态和行为的容器。actor线程顺序处理收到的消息。这样就让用户摆脱锁和线程管理的管理，因为一次只有一个线程对一个actor有效。但是，必须确保只有这个actor线程可以处理其内部状态。Actor的行为由receive函数定义，该函数包含收到的消息的处理逻辑。

Flink系统由3个分布式组件构成：JobClient，JobManager和TaskManager。JobClient从用户处得到Flink Job，并提交给JobManager。JobManager策划这个job的执行。首先，它分配所需的资源，主要就是TaskManagers上要执行的slot。

在资源分配之后，JobManager部署单独的任务到响应的TaskManager上。一旦收到一个任务，TaskManager产生一个线程用来执行这个任务。状态的改变，比如开始计算或者完成计算，将被发送回JobManager。基于这些状态的更新，JobManager将引导这个job的执行直到完成。一旦一个job被执行完，其结果将会被发送回JobClient。Job的执行图如下所示：

3、 异步VS同步消息

在任何地方，Flink尝试使用异步消息和通过futures来处理响应。Futures和很少的几个阻塞调用有一个超时时间，以防操作失败。这是为了防止死锁，当消息丢失或者分布式足觉crash。但是，如果在一个大集群或者慢网络的情况下，超时可能会使得情况更糟。因此，操作的超时时间可以通过“akka.timeout.timeout”来配置。

在两个actor可以通信之前，需要获取一个ActorRef。这个操作的查找同样需要一个超时。为了使得系统尽可能快速的失败，如果一个actor还没开始，超时时间需要被设置的比较小。为了以防经历查询超时，可以通过“akka.lookup.timeout”配置增加查询时间。

Akka的另一个特点是限制发送的最大消息大小。原因是它保留了同样数据大小的序列化buffer和不想浪费空间。如果你曾经遇到过传输失败，因为消息超过了最大大小，你可以增加“akka.framesize”配置来增加大小。

使用Akka

Akka系统的核心ActorSystem和Actor，若需构建一个Akka系统，首先需要创建ActorSystem，创建完ActorSystem后，可通过其创建Actor（注意：Akka不允许直接new一个Actor，只能通过 Akka 提供的某些 API 才能创建或查找 Actor，一般会通过 ActorSystem#actorOf和ActorContext#actorOf来创建 Actor），另外，我们只能通过ActorRef（Actor的引用，其对原生的 Actor 实例做了良好的封装，外界不能随意修改其内部状态）来与Actor进行通信。如下代码展示了如何配置一个Akka系统。

// 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();

1、 Actor路径

在Akka中，创建的每个Actor都有自己的路径，该路径遵循ActorSystem 的层级结构，大致如下：

1）本地路径

在上面代码中，本地Actor路径为
akka://sys/user/helloActor

含义如下：

• sys，创建的ActorSystem的名字；
• user，通过ActorSystem#actorOf和ActorContext#actorOf 方法创建的 Actor 都属于/user下，与/user对应的是/system， 其是系统层面创建的，与系统整体行为有关，在开发阶段并不需要对其过多关注
• helloActor，我们创建的HelloActor

2）远程路径

在上面代码中，远程Actor路径为 akka.tcp://sys@
l27.0.0.1:2020/user/remoteActor

含义如下：

• akka.tcp，远程通信方式为tcp；
• sys@127.0.0.1:2020，ActorSystem名字及远程主机ip和端口号。
• user，与本地的含义一样
• remoteActor，创建的远程Actor

2、 获取Actor

若提供了Actor的路径，可以通过路径获取到ActorRef，然后与之通信，代码如下所示：

ActorSystem system = ActorSystem.create("sys")；
ActorSelection as= system.actorSelection("/path/to/actor");
Timeout timeout =new Timeout(Duration.create(2, "seconds"));
Future<ActorRef> fu = as.resolveOne(timeout);
fu.onSuccess(newOnSuccess<ActorRef>() {
 @Override
public void onSuccess(ActorRef actor) {
 System.out.println("actor:" +actor);
 actor.tell("hello actor",ActorRef.noSender());
 }
},system.dispatcher());
fu.onFailure(newOnFailure() {
 @Override
public void onFailure(Throwable failure) {
 System.out.println("failure:" +failure);
 }
},system.dispatcher());

若需要与远端Actor通信，路径中必须提供ip:port。

Actor通信

Akka有两种核心的异步通信方式：tell和ask。

1、 Tell方式

当使用tell方式时，表示仅仅使用异步方式给某个Actor发送消息，无需等待Actor的响应结果，并且也不会阻塞后续代码的运行，如：

helloActor.tell("hellohelloActor", ActorRef.noSender());

其中：第一个参数为消息，它可以是任何可序列化的数据或对象，第二个参数表示发送者，通常来讲是另外一个 Actor 的引用， ActorRef.noSender()表示无发送者（（实际上是一个叫做deadLetters的Actor）。

2、 Ask方式

当我们需要从Actor获取响应结果时，可使用ask方法，ask方法会将返回结果包装在scala.concurrent.Future中，然后通过异步回调获取返回结果。如调用方：

// 异步发送消息给Actor，并获取响应结果
Future<Object> fu = Patterns.ask(printerActor, "hello helloActor", timeout);
fu.onComplete(newOnComplete<Object>() {
@Override
public void onComplete(Throwable failure, String success) throws Throwable {
if (failure != null) {
          System.out.println("failure is " + failure);
        }else {
          System.out.println("success is " + success);
        }
     }
},system.dispatcher());

HelloActor处理消息方法的代码大致如下：

private void handleMessage(Object object) {
if (objectinstanceof String) {
String str = (String)object;
 log.info("[HelloActor] message is {},sender is {}", str, getSender().path().toString());
// 给发送者发送消息
 getSender().tell(str, getSelf());
 }
 }

上面主要介绍了Akka中的ActorSystem、Actor，及与Actor的通信；Flink借此构建了其底层通信系统。

想要了解跟多关于大数据培训课程内容欢迎关注尚硅谷大数据培训，尚硅谷除了这些技术文章外还有免费的高质量大数据培训课程视频供广大学员下载学习。