flink中的rpc框架使用的akka。在本節並不詳細講述akka，而是就flink中rpc來講述akka的部分內容。本節，我從AkkaRpcActor.handleRpcInvocation方法講起。
看過hadoop、yarn、hive、hbase、presto的rpc框架，感覺flink的通信框架是最容易讓人繞暈的。雖然之前也看過一點spark中akka的通信，但現在早已忘得一干二凈。如今重拾akka通信，感覺還是挺復雜的。因此，這里特意拿出一節來講解。
1.這里首先要講述的是flink中關於心跳的rpc交互。這里也是akka中第一種遠程通信方式，也就是說通過tell方式異步傳輸。
這里我們從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后再無后續操作。
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的調用。
3.接着是akka的最后通信方式——阻塞返回。在flink中的對應的方法是AkkaRpcActor.sendSyncResponse(這里在flink中很少用到，因此我這里並沒有舉例)。
這里rpc調用方法的返回值為非CompletableFuture類型，前面的調用流程與上面講述的異步返回一樣，所不同的是，由於方法返回值類型為非CompletableFuture，因此，這里調用了CompletableFuture.get，這里會一直阻塞，直待該CompletableFuture的完成。這里的CompletableFuture其實就是通過FutureUtils.toJava實現了將scala中的future轉換為java中的CompletableFuture。也就是說，這里會一直等到遠程方法Promise.DefaultPromise的相應方法(success或failure)被調用，這里的阻塞才會被打斷。
好了，到這里為止，關於flink中應用akka完成其rpc通信框架的流程就結束了，感謝大家的關注。