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

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

總結(jié)

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

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