flink中的rpc框架使用的akka。在本節(jié)并不詳細講述akka，而是就flink中rpc來講述akka的部分內(nèi)容。本節(jié)，我從AkkaRpcActor.handleRpcInvocation方法講起。 　　看過hadoop、yarn、hive、hbase、presto的rpc框架，感覺flink的通信框架是最容易讓人繞暈的。雖然之前也看過一點spark中akka的通信，但現(xiàn)在早已忘得一干二凈。如今重拾akka通信，感覺還是挺復雜的。因此，這里特意拿出一節(jié)來講解。 　　1.這里首先要講述的是flink中關于心跳的rpc交互。這里也是akka中第一種遠程通信方式，也就是說通過tell方式異步傳輸。 如下圖所示，這里是我前幾天畫的《flink心跳》思維導圖的一部分，需要完整版加我微信——letusflyinthesky(有償出售，flink吹牛必備哦)。 　　這里我們從HeartbeatTarget.requestHeartbeat開始講。真正調(diào)用的是ResourceManager.registerTaskExecutorInternal方法中類型為HeartbeatTarget的匿名類，其內(nèi)部調(diào)用了taskExecutorGateway.heartbeatFromResourceManager。這里的taskExecutorGateway是一個代理類，其invocationHandler為AkkaInvocationHandler。因此，這里首先調(diào)用的是AkkaInvocationHandler.invoke，由于這里要調(diào)用的并非本地方法，因此接著調(diào)用了方法AkkaInvocationHandler.invokeRpc。在該方法中首先通過方法createRpcInvocationMessage封裝了發(fā)現(xiàn)taskmanager端的請求RemoteRpcInvocation，接著獲取了欲調(diào)用方法的返回值(這里的判斷是為了后面使用不同的akka通信方式)。我們這里的返回值為Void。然后調(diào)用了AkkaInvocationHandler.tell。這里的入?yún)⑹莿倓偡庋b的RemoteRpcInvocation，該方法內(nèi)部調(diào)用了ActorRef.tell。該actor就是taskmanager端的化生，發(fā)送了RemoteRpcInvocation(可序列化)。jobmanager端，也就是resourcemanager端的流程到這里就結束了，因為我們遠程調(diào)用的方法是無返回值的。 　　接著，我們來到taskmanager端，這里的AkkaRpcActor.onReceive接收到resourcemanager端發(fā)來的消息。根據(jù)類型的匹配，我們來到AkkaRpcActor.handleRpcMessage。由于這里的信息是RemoteRpcInvocation，實現(xiàn)了接口RpcInvocation，因此，我們來到AkkaRpcActor.handleRpcInvocation方法。這里首先調(diào)用方法lookupRpcMethod根據(jù)方法名獲取taskmanager端對應的方法，也就是TaskExecutor中對應的方法。接著，設置了其訪問屬性后，便開始反射調(diào)用。由于我們這里的方法返回值類型為Void，因此，在調(diào)用了TaskExecutor.heartbeatFromResourceManager后再無后續(xù)操作。 　　2.接著是akka中的第二種通信方式——異步返回。我這里的使用的是taskmanager向resourcemanager遠程注冊的例子來講解。 　　這里使用了akka的異步返回機制。如果對akka的異步返回不太熟悉的朋友，我推薦大家看一下http://sunxiang0918.cn/2016/01/10/Akka-in-JAVA-1/。這里一共有四篇文章，對于akka入門有極大裨益。另外，我會在下篇博客發(fā)布時，將整理的flink中關于akka的代碼發(fā)布到我的github上，到時大家可以參考一下。這里我配合思維導圖方便大家的理解。 　　從TaskExecutorToResourceManagerConnection.ResourceManagerRegistration.invokeRegistration講起。該方法內(nèi)部調(diào)用了resourceManager.registerTaskExecutor。這里的resourceManager實際類型是FencedAkkaInvocationHandler。FencedAkkaInvocationHandler繼承自AkkaInvocationHandler。這里的部分調(diào)用流程與上面的異步無返回類似，我就從其中不同的地方講起。由于我們這里的返回值類型為CompletableFuture<RegistrationResponse>，不是Void類型，因此，這里首先調(diào)用了FencedAkkaInvocationHandler.ask，接著調(diào)用了FencedAkkaInvocationHandler.fenceMessage將信息類型封裝為RemoteFencedMessage，接著調(diào)用AkkaInvocationHandler.ask。這里是比較復雜的地方。首先調(diào)用了Patterns.ask(ActorRef, message)，這里的ActorRef是resourcemanager端的化身，Patterns.ask是akka用于遠程異步調(diào)用的一種方式。其返回值為scala.concurrent.Future，也就是scala類型的Future。該類型有方法onComplete，作用是當該Future完成是，不論是拋出異常或返回值完成此未來時，調(diào)用該方法入?yún)⒅械暮瘮?shù)。這里我們通過FutureUtils.toJava將scala中的Future轉(zhuǎn)換為java中的CompletableFuture。得到CompletableFuture后，taskmanager端接著調(diào)用CompletableFuture.thenApply方法，內(nèi)部調(diào)用了返回值的deserializeValue方法，也就是獲取到遠程的序列化的返回值后，將其反序列化。由于我們這里rpc調(diào)用的方法返回值是CompletableFuture類型，因此這里并不阻塞，直接返回。 　　然后，我們來到resourcemanager端，這里的AkkaRpcActor.onReceive方法被調(diào)用(注意，這里的實際類型是FencedAkkaRpcActor)，由于傳入的類型為RemoteFencedMessage，這里接著調(diào)用了FencedAkkaRpcActor.handleRpcMessage。經(jīng)過幾個判斷后，這里調(diào)用了AkkaRpcActor.handleRpcMessage，此時，這里的入?yún)镽emoteFencedMessage.getPayload，也就是RemoteRpcInvocation。接下來的流程我在上面已經(jīng)提到，這里就不贅述了。所不同的是，我們這里的返回為類型為CompletableFuture，因此，這里接著會調(diào)用AkkaRpcActor.sendAsyncResponse。這里首先調(diào)用了方法——Patterns.pipe(promise.future(), getContext().dispatcher()).to(sender)，這里的promise是scala中的Promise.DefaultPromise類型，該方法的作用其實就是講java中的CompletableFuture轉(zhuǎn)換為scala中的類型DefaultPromise，畢竟，java中的CompletableFuture類型無法實現(xiàn)rpc。sendAsyncResponse方法的作用就是，當入?yún)syncResponse完成后，會調(diào)用Promise.DefaultPromise的相應方法(success或failure)被調(diào)用。此時，由于Patterns.pipe(promise.future(), getContext().dispatcher()).to(sender)已經(jīng)被調(diào)用，因此，taskmanager端調(diào)用Patterns.ask方法的返回的future為完成狀態(tài)，也就是調(diào)用了其onComplete。接著，在taskmanager端將返回值反序列化，完成異步rpc的調(diào)用。 　　3.接著是akka的最后通信方式——阻塞返回。在flink中的對應的方法是AkkaRpcActor.sendSyncResponse(這里在flink中很少用到，因此我這里并沒有舉例)。 　　這里rpc調(diào)用方法的返回值為非CompletableFuture類型，前面的調(diào)用流程與上面講述的異步返回一樣，所不同的是，由于方法返回值類型為非CompletableFuture，因此，這里調(diào)用了CompletableFuture.get，這里會一直阻塞，直待該CompletableFuture的完成。這里的CompletableFuture其實就是通過FutureUtils.toJava實現(xiàn)了將scala中的future轉(zhuǎn)換為java中的CompletableFuture。也就是說，這里會一直等到遠程方法Promise.DefaultPromise的相應方法(success或failure)被調(diào)用，這里的阻塞才會被打斷。 　　好了，到這里為止，關于flink中應用akka完成其rpc通信框架的流程就結束了，感謝大家的關注。另外，本人正在找成都大數(shù)據(jù)底層開發(fā)的工作，有推薦的朋友可以加我的微信交流(letusflyinthesky)，非誠勿擾。 ? ? ?

轉(zhuǎn)載于:https://www.cnblogs.com/letsfly/p/10853341.html

總結

以上是生活随笔為你收集整理的flink RPC(akka)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。