Android線程池源碼閱讀記錄介紹
今天面試被問到線程池如何復用線程的?當場就懵掉瞭…於是面試完畢就趕緊打開源碼看瞭看,在此記錄下:
我們都知道線程池的用法,一般就是先new一個ThreadPoolExecutor對象,再調用execute(Runnable runnable)傳入我們的Runnable,剩下的交給線程池處理就行瞭,於是這次我就從ThreadPoolExecutor的execute方法看起:
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */ int c = ctl.get(); //1.如果workerCountOf(c)即正在運行的線程數小於核心線程數,就執行addWork if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } //2.如果線程池還在運行狀態並且把任務添加到任務隊列成功 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); //3.如果線程池不在運行狀態並且從任務隊列移除任務成功,執行線程池飽和策略(默認直接拋出異常) if (! isRunning(recheck) && remove(command)) reject(command); //4.否則如果此時運行線程數==0,就直接調用addWork方法 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //5.如果2條件不成立,繼續判斷如果addWork返回false,執行線程池飽和策略 else if (!addWorker(command, false)) reject(command); }
大致過程就是如果核心線程未滿,則直接addWorker(該方法下面會再分析);如果核心線程已滿,則嘗試將任務加進消息隊列中,並再判斷如果此時運行線程數==0則調addWorker方法,否則不做任何處理(因為運行的線程處理完自己的任務後會去消息隊列中取任務來執行,下面會分析);如果任務隊列添加任務失敗,那麼直接addWorker(),如果addWorker返回false,執行飽和策略,下面我們就來看看addWorker裡面做瞭什麼
/** * @param firstTask the task the new thread should run first (or * null if none). Workers are created with an initial first task * (in method execute()) to bypass queuing when there are fewer * than corePoolSize threads (in which case we always start one), * or when the queue is full (in which case we must bypass queue). * Initially idle threads are usually created via * prestartCoreThread or to replace other dying workers. * * @param core if true use corePoolSize as bound, else * maximumPoolSize. (A boolean indicator is used here rather than a * value to ensure reads of fresh values after checking other pool * state). * @return true if successful */ private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c); //1.如果正在運行的線程數大於corePoolSize 或 maximumPoolSize(core代表以核心線程數還是最大線程數為邊界),return false,表示addWorker失敗 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //2.否則將運行線程數+1,並跳出這個for循環 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { //3.創建一個Worker對象,傳入我們的runnable w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { //4.開始啟動線程 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker. */ public void run() { runWorker(this); } final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //1.當firstTask不為空或getTask不為空時一直循環 while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { //2.執行任務 task.run(); } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }
可以看到addWorker方法主要就是先判斷正在運行線程數是否超過瞭最大線程數(具體根據邊界取),如果未超過則創建一個worker對象,其中firstTask是我們傳入的Runnable,當然根據上面的execute方法可知當4條件滿足時,傳入的firstTask是null,Thread是用ThreadFactory創建的線程,傳入的Runnable是Worker自己,最後開啟線程,於是執行Worker這裡的run、runWorker方法,在runWorker方法裡,開啟一個while循環,當firstTask不為空或getTask不為空時,執行task,下面我們接著看看getTask裡面做瞭什麼:
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // Are workers subject to culling? //1.會不會淘汰空閑線程 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //2.return null意味著回收一個Worker即淘汰一個線程 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { //3.等待指定時間 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
可以看1、2註釋,allowCoreThreadTimeOut代表存活一定時間是否對核心線程有效(默認為false),先看它為ture的情況,此時不管是核心線程還是非核心線程在3處都會等待一定時間(就是我們傳入的線程保活時間),等待時間內如果從任務隊列取到任務,則返回執行,否則timeout為true,繼續走到2,由於(timed && timedOut)和workQueue.isEmpty()均為true,返回null,代表回收一個線程;如果allowCoreThreadTimeOut為false,代表不回收核心線程,此時如果在3處沒有取到任務,繼續執行到2處,隻有當wc > corePoolSize或wc > maximumPoolSize時才會執行return null,否則一直循環,相當於該線程一直處於運行狀態,直到從任務隊列拿到新的任務
到此這篇關於Android線程池源碼閱讀記錄介紹的文章就介紹到這瞭,更多相關Android線程池內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- 一篇文章徹底搞懂jdk8線程池
- Android開發中線程池源碼解析
- java線程池中Worker線程執行流程原理解析
- Java線程池ThreadPoolExecutor源碼深入分析
- 詳解Java線程池是如何重復利用空閑線程的