❶ Spark源碼分析之SparkSubmit的流程
本文主要對SparkSubmit的任務提交流程源碼進行分析。 Spark源碼版本為2.3.1。
首先閱讀一下啟動腳本,看看首先載入的是哪個類,我們看一下 spark-submit 啟動腳本中的具體內容。
可以看到這里載入的類是org.apache.spark.deploy.SparkSubmit,並且把啟動相關的參數也帶過去了。下面我們跟一下源碼看看整個流程是如何運作的...
SparkSubmit的main方法如下
這里我們由於我們是提交作業,所有會走上面的submit(appArgs, uninitLog)方法
可以看到submit方法首先會准備任務提交的環境,調用了prepareSubmitEnvironment,該方法會返回四元組,該方法中會調用doPrepareSubmitEnvironment,這里我們重點注意 childMainClass類具體是什麼 ,因為這里涉及到後面啟動我們主類的過程。
以下是doPrepareSubmitEnvironment方法的源碼...
可以看到該方法首先是解析相關的參數,如jar包,mainClass的全限定名,系統配置,校驗一些參數,等等,之後的關鍵點就是根據我們 deploy-mode 參數來判斷是如何運行我們的mainClass,這里主要是通過childMainClass這個參數來決定下一步首先啟動哪個類。
childMainClass根據部署模型有不同的值:
之後該方法會把准備好的四元組返回,我們接著看之前的submit方法
可以看到這里最終會調用doRunMain()方法去進行下一步。
doRunMain的實現如下...
doRunMain方法中會判斷是否需要一個代理用戶,然後無論需不需要都會執行runMain方法,我們接下來看看runMain方法是如何實現的。
這里我們只假設以集群模式啟動,首先會載入類,將我們的childMainClass載入為位元組碼對象mainClass ,然後將mainClass 映射成SparkApplication對象,因為我們以集群模式啟動,那麼上一步返回四元組中的childMainClass的參數為ClientApp的全限定名,而這里會調用app實例的start方法因此,這里最終調用的是ClientApp的start方法。
ClientApp的start方法如下...
可以看到這里和之前我們的master啟動流程有些相似。
可以參考我上一篇文章 Spark源碼分析之Master的啟動流程 對這一流程加深理解。
首先是准備rpcEnv環境,之後通過master的地址獲取masterEndpoints端點相關信息,因為這里運行start方法時會將之前配置的相關參數都傳進來,之後就會通過rpcEnv注冊相關clientEndPoint端點信息,同時需要注意,這里會把masterEndpoints端點信息也作為構造ClientEndpoint端點的參數,也就是說這個ClientEndpoint會和masterEndpoints通信。
而在我上一篇文章中說過,只要是setupEndpoint方法被調用,一定會調用相關端點的的onStart方法,而這會調用clientEndPoint的onStart方法。
ClientEndPoint類中的onStart方法會匹配launch事件。源碼如下
onStart中匹配我們的launch的過程,這個過程是啟動driverWrapper的過程,可以看到上面源碼中封裝了mainClass ,該參數對應DriverWrapper類的全限定名,之後將mainClass封裝到command中,然後封裝到driverDescription中,向Master申請啟動Driver。
這個過程會向Mster發送消息,是通過rpcEnv來實現發射消息的,而這里就涉及到outbox信箱,會調用postToOutbox方法,向outbox信箱中添加消息,然後通過TransportClient的send或sendRpc方法發送消息。發件箱以及發送過程是在同一個線程中進行。
而細心的同學會注意到這里調用的方法名為SendToMasterAndForwardReply,見名之意,發送消息到master並且期待回應。
下面是rpcEnv來實現向遠端發送消息的一個調用流程,最終會通過netty中的TransportClient來寫出。
之後,Master端會觸發receiveAndReply函數,匹配RequestSubmitDriver樣例類,完成模式匹配執行後續流程。
可以看到這里首先將Driver信息封裝成DriverInfo,然後添加待調度列表waitingDrivers中,然後調用通用的schele函數。
由於waitingDrivers不為空,則會走LaunchDriver的流程,當前的application申請資源,這時會向worker發送消息,觸發Worker的receive方法。
Worker的receive方法中,當Worker遇到LaunchDriver指令時,創建並啟動一個DriverRunner,DriverRunner啟動一個線程,非同步的處理Driver啟動工作。這里說啟動的Driver就是剛才說的org.apache.spark.deploy.worker.DriverWrapper
可以看到上面在DriverRunner中是開辟線程非同步的處理Driver啟動工作,不會阻塞主進程的執行,而prepareAndRunDriver方法中最終調用 runDriver..
runDriver中主要先做了一些初始化工作,接著就開始啟動driver了。
上述Driver啟動工作主要分為以下幾步:
下面我們直接看DriverWrapper的實現
DriverWrapper,會創建了一個RpcEndpoint與RpcEnv,RpcEndpoint為WorkerWatcher,主要目的為監控Worker節點是否正常,如果出現異常就直接退出,然後當前的ClassLoader載入userJar,同時執行userMainClass,在執行用戶的main方法後關閉workerWatcher。
以上就是SparkSubmit的流程,下一篇我會對SparkContext的源碼進行解析。
歡迎關注...
❷ android 中的「子線程」解析
Android 中線程可分為 主線程 和 子線程 兩類,其中主線程也就是 UI線程 ,它的主要這作用就是運行四大組件、處理界面交互。子線程則主要是處理耗時任務,也是我們要重點分析的。
首先 Java 中的各種線程在 Android 里是通用的,Android 特有的線程形態也是基於 Java 的實現的,所以有必要先簡單的了解下 Java 中的線程,本文主要包括以下內容:
在 Java 中要創建子線程可以直接繼承 Thread 類,重寫 run() 方法:
或者實現 Runnable 介面,然後用Thread執行Runnable,這種方式比較常用:
簡單的總結下:
Callable 和 Runnable 類似,都可以用來處理具體的耗時任務邏輯的,但是但具體的差別在哪裡呢?看一個小例子:
定義 MyCallable 實現了 Callable 介面,和之前 Runnable 的 run() 方法對比下, call() 方法是有返回值的哦,泛型就是返回值的類型:
一般會通過線程池來執行 Callable (線程池相關內容後邊會講到),執行結果就是一個 Future 對象:
可以看到,通過線程池執行 MyCallable 對象返回了一個 Future 對象,取出執行結果。
Future 是一個介面,從其內部的方法可以看出它提供了取消任務(有坑!!!)、判斷任務是否完成、獲取任務結果的功能:
Future 介面有一個 FutureTask 實現類,同時 FutureTask 也實現了 Runnable 介面,並提供了兩個構造函數:
用 FutureTask 一個參數的構造函數來改造下上邊的例子:
FutureTask 內部有一個 done() 方法,代表 Callable 中的任務已經結束,可以用來獲取執行結果:
所以 Future + Callable 的組合可以更方便的獲取子線程任務的執行結果,更好的控制任務的執行,主要的用法先說這么多了,其實 AsyncTask 內部也是類似的實現!
注意, Future 並不能取消掉運行中的任務,這點在後邊的 AsyncTask 解析中有提到。
Java 中線程池的具體的實現類是 ThreadPoolExecutor ,繼承了 Executor 介面,這些線程池在 Android 中也是通用的。使用線程池的好處:
常用的構造函數如下:
一個常規線程池可以按照如下方式來實現:
執行任務:
基於 ThreadPoolExecutor ,系統擴展了幾類具有新特性的線程池:
線程池可以通過 execute() 、 submit() 方法開始執行任務,主要差別從方法的聲明就可以看出,由於 submit() 有返回值,可以方便得到任務的執行結果:
要關閉線程池可以使用如下方法:
IntentService 是 Android 中一種特殊的 Service,可用於執行後台耗時任務,任務結束時會自動停止,由於屬於系統的四大組件之一,相比一般線程具有較高的優先順序,不容易被殺死。用法和普通 Service 基本一致,只需要在 onHandleIntent() 中處理耗時任務即可:
至於 HandlerThread,它是 IntentService 內部實現的重要部分,細節內容會在 IntentService 源碼中說到。
IntentService 首次創建被啟動的時候其生命周期方法 onCreate() 會先被調用,所以我們從這個方法開始分析:
這里出現了 HandlerThread 和 ServiceHandler 兩個類,先搞明白它們的作用,以便後續的分析。
首先看 HandlerThread 的核心實現:
首先它繼承了 Thread 類,可以當做子線程來使用,並在 run() 方法中創建了一個消息循環系統、開啟消息循環。
ServiceHandler 是 IntentService 的內部類,繼承了 Handler,具體內容後續分析:
現在回過頭來看 onCreate() 方法主要是一些初始化的操作, 首先創建了一個 thread 對象,並啟動線程,然後用其內部的 Looper 對象 創建一個 mServiceHandler 對象,將子線程的 Looper 和 ServiceHandler 建立了綁定關系,這樣就可以使用 mServiceHandler 將消息發送到子線程去處理了。
生命周期方法 onStartCommand() 方法會在 IntentService 每次被啟動時調用,一般會這里處理啟動 IntentService 傳遞 Intent 解析攜帶的數據:
又調用了 start() 方法:
就是用 mServiceHandler 發送了一條包含 startId 和 intent 的消息,消息的發送還是在主線程進行的,接下來消息的接收、處理就是在子線程進行的:
當接收到消息時,通過 onHandleIntent() 方法在子線程處理 intent 對象, onHandleIntent() 方法執行結束後,通過 stopSelf(msg.arg1) 等待所有消息處理完畢後終止服務。
為什麼消息的處理是在子線程呢?這里涉及到 Handler 的內部消息機制,簡單的說,因為 ServiceHandler 使用的 Looper 對象就是在 HandlerThread 這個子線程類里創建的,並通過 Looper.loop() 開啟消息循環,不斷從消息隊列(單鏈表)中取出消息,並執行,截取 loop() 的部分源碼:
dispatchMessage() 方法間接會調用 handleMessage() 方法,所以最終 onHandleIntent() 就在子線程中劃線執行了,即 HandlerThread 的 run() 方法。
這就是 IntentService 實現的核心,通過 HandlerThread + Hanlder 把啟動 IntentService 的 Intent 從主線程切換到子線程,實現讓 Service 可以處理耗時任務的功能!
AsyncTask 是 Android 中輕量級的非同步任務抽象類,它的內部主要由線程池以及 Handler 實現,在線程池中執行耗時任務並把結果通過 Handler 機制中轉到主線程以實現UI操作。典型的用法如下:
從 Android3.0 開始,AsyncTask 默認是串列執行的:
如果需要並行執行可以這么做:
AsyncTask 的源碼不多,還是比較容易理解的。根據上邊的用法,可以從 execute() 方法開始我們的分析:
看到 @MainThread 註解了嗎?所以 execute() 方法需要在主線程執行哦!
進而又調用了 executeOnExecutor() :
可以看到,當任務正在執行或者已經完成,如果又被執行會拋出異常!回調方法 onPreExecute() 最先被執行了。
傳入的 sDefaultExecutor 參數,是一個自定義的串列線程池對象,所有任務在該線程池中排隊執行:
可以看到 SerialExecutor 線程池僅用於任務的排隊, THREAD_POOL_EXECUTOR 線程池才是用於執行真正的任務,就是我們線程池部分講到的 ThreadPoolExecutor :
再回到 executeOnExecutor() 方法中,那麼 exec.execute(mFuture) 就是觸發線程池開始執行任務的操作了。
那 executeOnExecutor() 方法中的 mWorker 是什麼? mFuture 是什麼?答案在 AsyncTask 的構造函數中:
原來 mWorker 是一個 Callable 對象, mFuture 是一個 FutureTask 對象,繼承了 Runnable 介面。所以 mWorker 的 call() 方法會在 mFuture 的 run() 方法中執行,所以 mWorker 的 call() 方法在線程池得到執行!
同時 doInBackground() 方法就在 call() 中方法,所以我們自定義的耗時任務邏輯得到執行,不就是我們第二部分講的那一套嗎!
doInBackground() 的返回值會傳遞給 postResult() 方法:
就是通過 Handler 將最終的耗時任務結果從子線程發送到主線程,具體的過程是這樣的, getHandler() 得到的就是 AsyncTask 構造函數中初始化的 mHandler , mHander 又是通過 getMainHandler() 賦值的:
可以在看到 sHandler 是一個 InternalHandler 類對象:
所以 getHandler() 就是在得到在主線程創建的 InternalHandler 對象,所以
就可以完成耗時任務結果從子線程到主線程的切換,進而可以進行相關UI操作了。
當消息是 MESSAGE_POST_RESULT 時,代表任務執行完成, finish() 方法被調用:
如果任務沒有被取消的話執行 onPostExecute() ,否則執行 onCancelled() 。
如果消息是 MESSAGE_POST_PROGRESS , onProgressUpdate() 方法被執行,根據之前的用法可以 onProgressUpdate() 的執行需要我們手動調用 publishProgress() 方法,就是通過 Handler 來發送進度數據:
進行中的任務如何取消呢?AsyncTask 提供了一個 cancel(boolean mayInterruptIfRunning) ,參數代表是否中斷正在執行的線程任務,但是呢並不靠譜, cancel() 的方法注釋中有這么一段:
大致意思就是調用 cancel() 方法後, onCancelled(Object) 回調方法會在 doInBackground() 之後被執行而 onPostExecute() 將不會被執行,同時你應該 doInBackground() 回調方法中通過 isCancelled() 來檢查任務是否已取消,進而去終止任務的執行!
所以只能自己動手了:
AsyncTask 整體的實現流程就這些了,源碼是最好的老師,自己跟著源碼走一遍有些問題可能就豁然開朗了!
❸ Android源碼追蹤—android:onClick
之前對源碼的閱讀,總是用時一通亂七八糟的跳轉,以學會使用為目的;過了一段時間,就忘記了,因此打算將一些源碼的閱讀經歷記錄下來,也通過敲一遍的過程,加深理解。
最開始,用一個比較簡單的例子來小試牛刀吧
對於View(Button、TextView等)的點擊事件,常用的寫法是通過 findViewById 獲取View的實例,然後通過 setOnClickListener 設置監聽事件,比如我們有如下Button控制項。
設置點擊事件(假設在Activity中)
但是還有一種寫法是在xml布局中通過android:onClick屬性直接指定點擊執行的函數。
【思考】
首先我們知道諸如 android:xxx 之類的屬性是會在某個attrs文件中定義的,此處的 android:onClick 是View的屬性,定義在如下文件中。
在View的構造函數中,會解析出此屬性的值。
看這里, 如果變數handlerName不為空,就會為此View設置點擊事件了 ,這個handlerName就是onClick屬性的值doSubmit,但這個點擊事件,並不是我們所熟悉的OnClickListener。
進一步看看這個 DeclaredOnClickListener 類
DeclaredOnClickListener 實現了 OnClickListener ,其中重點是參數 mResolvedMethod 和 mResolvedContext 。
在onClick事件中最終通過反射 mResolvedMethod.invoke(mResolvedContext, v); 執行了doSubmit方法。
doSubmit的訪問許可權是否可以設置為private呢?
答案:不可以,因為源碼中沒有調用 mMethod.setAccessible(true); 注入所有修飾符。
其實在onClick屬性的注釋中就已經說明了。