flink中的rpc框架使用的akka。在本节并不详细讲述akka,而是就flink中rpc来讲述akka的部分内容。本节,我从AkkaRpcActor.handleRpcInvocation方法讲起。
看过hadoop、yarn、hive、hbase、presto的rpc框架,感觉flink的通信框架是最容易让人绕晕的。虽然之前也看过一点spark中akka的通信,但现在早已忘得一干二净。如今重拾akka通信,感觉还是挺复杂的。因此,这里特意拿出一节来讲解。
1.这里首先要讲述的是flink中关于心跳的rpc交互。这里也是akka中第一种远程通信方式,也就是说通过tell方式异步传输。
如下图所示,这里是我前几天画的《flink心跳》思维导图的一部分,需要完整版加我微信——letusflyinthesky(有偿出售,flink吹牛必备哦)。
这里我们从HeartbeatTarget.requestHeartbeat开始讲。真正调用的是ResourceManager.registerTaskExecutorInternal方法中
类型为HeartbeatTarget的匿名类,其内部调用了taskExecutorGateway.heartbeatFromResourceManager。这里的taskExecutorGateway是一个代理类,其invocationHandler为AkkaInvocationHandler。因此,这里首先调用的是AkkaInvocationHandler.invoke,由于这里要调用的并非本地方法,因此接着调用了方法AkkaInvocationHandler.invokeRpc。在该方法中首先通过方法createRpcInvocationMessage封装了发现taskmanager端的请求RemoteRpcInvocation,接着获取了欲调用方法的返回值(这里的判断是为了后面使用不同的akka通信方式)。我们这里的返回值为Void。然后调用了AkkaInvocationHandler.tell。这里的入参是刚刚封装的RemoteRpcInvocation,该方法内部调用了ActorRef.tell。该actor就是taskmanager端的化生,发送了RemoteRpcInvocation(可序列化)。jobmanager端,也就是resourcemanager端的流程到这里就结束了,因为我们远程调用的方法是无返回值的。
接着,我们来到taskmanager端,这里的AkkaRpcActor.onReceive接收到resourcemanager端发来的消息。根据类型的匹配,我们来到AkkaRpcActor.handleRpcMessage。由于这里的信息是RemoteRpcInvocation,实现了接口RpcInvocation,因此,我们来到AkkaRpcActor.handleRpcInvocation方法。这里首先调用方法lookupRpcMethod根据方法名获取taskmanager端对应的方法,也就是TaskExecutor中对应的方法。接着,设置了其访问属性后,便开始反射调用。由于我们这里的方法返回值类型为Void,因此,在调用了TaskExecutor.heartbeatFromResourceManager后再无后续操作。
![](https://img2018.cnblogs.com/blog/1365945/201905/1365945-20190512192811001-56592462.png)
2.接着是akka中的第二种通信方式——异步返回。我这里的使用的是taskmanager向resourcemanager远程注册的例子来讲解。
这里使用了akka的异步返回机制。如果对akka的异步返回不太熟悉的朋友,我推荐大家看一下http://sunxiang0918.cn/2016/01/10/Akka-in-JAVA-1/。这里一共有四篇文章,对于akka入门有极大裨益。另外,我会在下篇博客发布时,将整理的flink中关于akka的代码发布到我的github上,到时大家可以参考一下。这里我配合思维导图方便大家的理解。
从TaskExecutorToResourceManagerConnection.ResourceManagerRegistration.invokeRegistration讲起。该方法内部调用了resourceManager.registerTaskExecutor。这里的resourceManager实际类型是FencedAkkaInvocationHandler。FencedAkkaInvocationHandler继承自AkkaInvocationHandler。这里的部分调用流程与上面的异步无返回类似,我就从其中不同的地方讲起。由于我们这里的返回值类型为CompletableFuture<RegistrationResponse>,不是Void类型,因此,这里首先调用了FencedAkkaInvocationHandler.ask,接着调用了FencedAkkaInvocationHandler.fenceMessage将信息类型封装为RemoteFencedMessage,接着调用AkkaInvocationHandler.ask。这里是比较复杂的地方。首先调用了Patterns.ask(ActorRef, message),这里的ActorRef是resourcemanager端的化身,Patterns.ask是akka用于远程异步调用的一种方式。其返回值为scala.concurrent.Future,也就是scala类型的Future。该类型有方法onComplete,作用是当该Future完成是,不论是抛出异常或返回值完成此未来时,调用该方法入参中的函数。这里我们通过FutureUtils.toJava将scala中的Future转换为java中的CompletableFuture。得到CompletableFuture后,taskmanager端接着调用CompletableFuture.thenApply方法,内部调用了返回值的deserializeValue方法,也就是获取到远程的序列化的返回值后,将其反序列化。由于我们这里rpc调用的方法返回值是CompletableFuture类型,因此这里并不阻塞,直接返回。
然后,我们来到resourcemanager端,这里的AkkaRpcActor.onReceive方法被调用(注意,这里的实际类型是FencedAkkaRpcActor),由于传入的类型为RemoteFencedMessage,这里接着调用了FencedAkkaRpcActor.handleRpcMessage。经过几个判断后,这里调用了AkkaRpcActor.handleRpcMessage,此时,这里的入参为RemoteFencedMessage.getPayload,也就是RemoteRpcInvocation。接下来的流程我在上面已经提到,这里就不赘述了。所不同的是,我们这里的返回为类型为CompletableFuture,因此,这里接着会调用AkkaRpcActor.sendAsyncResponse。这里首先调用了方法——Patterns.pipe(promise.future(), getContext().dispatcher()).to(sender),这里的promise是scala中的Promise.DefaultPromise类型,该方法的作用其实就是讲java中的CompletableFuture转换为scala中的类型DefaultPromise,毕竟,java中的CompletableFuture类型无法实现rpc。sendAsyncResponse方法的作用就是,当入参asyncResponse完成后,会调用Promise.DefaultPromise的相应方法(success或failure)被调用。此时,由于Patterns.pipe(promise.future(), getContext().dispatcher()).to(sender)已经被调用,因此,taskmanager端调用Patterns.ask方法的返回的future为完成状态,也就是调用了其onComplete。接着,在taskmanager端将返回值反序列化,完成异步rpc的调用。
![](https://img2018.cnblogs.com/blog/1365945/201905/1365945-20190512192910841-637608496.png)
这里rpc调用方法的返回值为非CompletableFuture类型,前面的调用流程与上面讲述的异步返回一样,所不同的是,由于方法返回值类型为非CompletableFuture,因此,这里调用了CompletableFuture.get,这里会一直阻塞,直待该CompletableFuture的完成。这里的CompletableFuture其实就是通过FutureUtils.toJava实现了将scala中的future转换为java中的CompletableFuture。也就是说,这里会一直等到远程方法Promise.DefaultPromise的相应方法(success或failure)被调用,这里的阻塞才会被打断。
好了,到这里为止,关于flink中应用akka完成其rpc通信框架的流程就结束了,感谢大家的关注。另外,本人正在找成都大数据底层开发的工作,有推荐的朋友可以加我的微信交流(letusflyinthesky),非诚勿扰。