提交流程呼叫的關鍵方法鏈

使用者編寫的程式邏輯需要提交給Flink才能得到執行。本文來探討一下客戶程式如何提交給Flink。鑑於使用者將自己利用Flink的API編寫的邏輯打成相應的應用程式包（比如Jar）然後提交到一個目標Flink叢集上去執行是比較主流的使用場景，因此我們的分析也基於這一場景進行。

Flink的API針對不同的執行環境有不同的Environment物件，這裡我們主要基於常用的RemoteStreamEnvironment和RemoteEnvironment進行分析

在前面我們談到了Flink中實現了“惰性求值”，只有當最終呼叫execute方法時，才會“真正”開始執行。因此，execute

其原始碼位於org.apache.flink.streaming.api.environment.RemoteStreamEnvironment

首先，我們來看一下其execute方法觸發的關鍵方法呼叫鏈示意圖：

根據上圖的呼叫鏈，我們針對這些關鍵方法進行剖析，當然一些細節性的內容我們可能會暫時略過，這樣可以保證主路徑一直都很清晰。

getStreamGraph方法用於獲得一個StreamGraph的例項，該例項表示流的完整的拓撲結構並且包含了生成JobGraph所必要的相關資訊（包含了source、sink的集合以及這些在圖中的“節點”抽象化的表示、一些虛擬的對映關係、執行和檢查點的配置等）。

獲得StreamGraph之後，通過呼叫executeRemotely方法進行遠端執行。該方法首先根據獲取到的使用者程式包的路徑以及類路徑建立載入使用者程式碼的類載入器：

ClassLoader usercodeClassLoader = JobWithJars.buildUserCodeClassLoader(
jarFiles, globalClasspaths,   getClass().getClassLoader());

緊接著根據配置構建Client物件（Client物件是真正跟JobManager對接的內部代理）：

Client client;
try {   
    client
 
 = new Client(configuration);   
    client.setPrintStatusDuringExecution(getConfig().isSysoutLoggingEnabled());
}catch (Exception e) {
   throw new ProgramInvocationException("Cannot establish connection to JobManager: " + e.getMessage(), e);
}

後面的事情就此被Client接管：

try {   
    return client.runBlocking(streamGraph, jarFiles, globalClasspaths, usercodeClassLoader);
}catch (ProgramInvocationException e) {   
    throw e;
}catch (Exception e) {   
    String term = e.getMessage() == null ? "." : (": " + e.getMessage());   
    throw new ProgramInvocationException("The program execution failed" + term, e);
}finally {   
    client.shutdown();
}

client物件呼叫了runBlocking以阻塞式的行為“執行”使用者程式並等待返回JobExecutionResult物件作為Job的執行結果。執行完成，最終在finally塊中，呼叫shutdown方法關閉並釋放資源。

runBlocking被呼叫後，呼叫鏈跳轉到Client類中。為了適配多種提交方式以及執行模式，runBlocking方法有著非常多的過載。在當前的遠端執行環境下，runBlocking在多個過載方法之間跳轉的過程中，會呼叫getJobGraph方法獲得JobGraph的例項。JobGraph表示Flink dataflow 程式，它將會被JobManager所理解並接收。在某個Job被提交給JobManager之前，通過Flink提供的高層次的API都將會被轉化為JobGraph表示。關於如何獲得JobGraph的實現，我們後面會進行剖析。這裡，讓我們忽視這些細節，進入下一個關鍵方法。

runBlocking_1其實是runBlocking方法的過載，這裡加一個字尾標識，只是為了跟上面的runBlocking進行區別。runBlocking_1方法中，首先利用LeaderRetrievalUtils建立了LeaderRetrievalService這一服務物件：

LeaderRetrievalService leaderRetrievalService;
try {   
    leaderRetrievalService = LeaderRetrievalUtils.createLeaderRetrievalService(config);
} catch (Exception e) {   
    throw new ProgramInvocationException("Could not create the leader retrieval service.", e);
}

顧名思義，LeaderRetrievalService在Flink中提供查詢主節點的服務。它會根據Flink的配置資訊（主要是recovery.mode來判斷基於哪種恢復機制來建立該服務。當前有兩種模式：一種是Standalone的獨立執行模式；另一種是基於Zookeeper的高可用模式）。Flink提供了一個稱之為LeaderRetrievalListener的回撥介面來獲得主節點的資訊。接下來，就是呼叫JobClient的submitJobAndWait方法將產生的JobGraph以及主節點查詢的服務物件等相關資訊提交給JobManager並等待返回結果：

try {   
    this.lastJobID = jobGraph.getJobID();   
    return JobClient.submitJobAndWait(actorSystem, leaderRetrievalService, jobGraph, 
                                        timeout, printStatusDuringExecution, classLoader);
} catch (JobExecutionException e) {   
    throw new ProgramInvocationException("The program execution failed: " + e.getMessage(), e);
}

上面的submitJobAndWait方法的第一個引數actorSystem是ActorSystem的例項。在構造Client物件時建立，在Job提交併獲得返回結果後通過呼叫Client的shutdown方法關閉：

public void shutdown() {   
    if (!this.actorSystem.isTerminated()) {      
        this.actorSystem.shutdown();      
        this.actorSystem.awaitTermination();   
    }
}

該方法的呼叫見上面executeRemotely方法的程式碼段的finally語句塊。

JobClient的出現可能會讓你產生疑惑——它跟Client是什麼關係？作用是什麼？下面這幅示意圖可以用來解釋這些疑問：

上面這幅圖展示了Client物件與其他幾個物件的關係。JobClient在其中起到了“橋接”作用，它在基於API的程式設計層面上橋接了同步的方法呼叫和非同步的訊息通訊。更具體得說，JobClient可以看做是一個“靜態類”提供了一些靜態方法，這裡我們主要關注上面的submitJobAndWait方法，該方法內部封裝了Actor之間的非同步通訊（具體的通訊物件是JobClientActor，它負責跟JobManager的ActorSystem的Actor進行通訊），並以阻塞的形式返回結果。而Client只需呼叫JobClient的這些方法，而無需關注其內部是如何實現的。

通過呼叫JobClient的靜態方法submitJobAndWait，會觸發基於Akka的Actor之間的訊息通訊來完成後續的提交JobGraph的動作。JobClient提交Job的基於訊息互動的抽象示意圖如下：

總體來說這裡總共有兩個ActorSystem，一個歸屬於Client，另一個歸屬於JobManager。在submitJobAndWait方法中，其首先會建立一個JobClientActor的ActorRef：

ActorRef jobClientActor = actorSystem.actorOf(jobClientActorProps);

然後向其發起一個SubmitJobAndWait訊息，該訊息將JobGraph的例項提交給jobClientActor。該訊息的發起模式是ask，它表示需要一個應答訊息。

JobClient向JobClientActor傳送訊息的程式碼段如下所示：

Future<Object> future = Patterns.ask(jobClientActor,      
                                     new JobClientMessages.SubmitJobAndWait(jobGraph),      
                                     new Timeout(AkkaUtils.INF_TIMEOUT()));
answer = Await.result(future, AkkaUtils.INF_TIMEOUT());

JobClient會阻塞等待該future返回結果。在得到返回結果answer之後，先進行解析判斷它是Job被成功執行返回的結果還是失敗返回的結果。

小結

至此，Client提交Streaming Job的關鍵方法呼叫路徑已梳理完成。這裡為了突出主路線，同時避免被太多的實現細節干擾，我們暫時忽略了一些重要資料結構和關鍵概念的解讀。不過，後續我們會對它們進行分析。

微信掃碼關注公眾號：Apache_Flink

QQ掃碼關注QQ群：Apache Flink學習交流群（123414680）