偷偷摘套内射激情视频,久久精品99国产国产精,中文字幕无线乱码人妻,中文在线中文a,性爽19p

Java線程池架構(gòu)(一)原理和源碼解析

開發(fā) 后端
我們這里將做源碼解析以及反饋到原理 上,Executors工具可以創(chuàng)建普通的線程池以及schedule調(diào)度任務(wù)的調(diào)度池。

在前面介紹JUC的文章中,提到了關(guān)于線程池Execotors的創(chuàng)建介紹,在文章:《java之JUC系列-外部Tools》中第一部分有詳細(xì)的說明,請參閱;

文章中其實說明了外部的使用方式,但是沒有說內(nèi)部是如何實現(xiàn)的,為了加深對實現(xiàn)的理解,在使用中可以放心,我們這里將做源碼解析以及反饋到原理 上,Executors工具可以創(chuàng)建普通的線程池以及schedule調(diào)度任務(wù)的調(diào)度池,其實兩者實現(xiàn)上還是有一些區(qū)別,但是理解了ThreadPoolExecutor,在看ScheduledThreadPoolExecutor就非常輕松了,后面的文章中也會專門介紹這塊,但是需要先看這篇文章。

 

使用Executors最常用的莫過于是使用:Executors.newFixedThreadPool(int)這個方法,因為它既可以限制數(shù)量,而且線程用完后不會一直被cache??;那么就通過它來看看源碼,回過頭來再看其他構(gòu)造方法的區(qū)別:

在《java之JUC系列-外部Tools》文章中提到了構(gòu)造方法,為了和本文對接,再貼下代碼。

  1. public static ExecutorService <strong>newFixedThreadPool</strong>(int nThreads) { 
  2.         return new ThreadPoolExecutor(nThreads, nThreads, 
  3.                                       0L, TimeUnit.MILLISECONDS, 
  4.                                       new LinkedBlockingQueue()); 

其實你可以自己new一個ThreadPoolExecutor,來達(dá)到自己的參數(shù)可控的程度,例如,可以將LinkedBlockingQueue換成其它的(如:SynchronousQueue),只是可讀性會降低,這里只是使用了一種設(shè)計模式。

我們現(xiàn)在來看看ThreadPoolExecutor的源碼是怎么樣的,也許你剛開始看他的源碼會很痛苦,因為你不知道作者為什么是這樣設(shè)計的,所以本文就我看到的思想會給你做一個介紹,此時也許你通過知道了一些作者的思想,你也許就知道應(yīng)該該如何去操作了。

這里來看下構(gòu)造方法中對那些屬性做了賦值:

源碼段1:

  1. public ThreadPoolExecutor(int corePoolSize, 
  2.                            int maximumPoolSize, 
  3.                            long keepAliveTime, 
  4.                            TimeUnit unit, 
  5.                            BlockingQueue workQueue, 
  6.                            ThreadFactory threadFactory, 
  7.                            RejectedExecutionHandler handler) { 
  8.      if (corePoolSize < 0 || 
  9.          maximumPoolSize <= 0 || 
  10.          maximumPoolSize < corePoolSize || 
  11.          keepAliveTime < 0
  12.          throw new IllegalArgumentException(); 
  13.     if (workQueue == null || threadFactory == null || handler == null
  14.           throw new NullPointerException(); 
  15.       this.corePoolSize = corePoolSize; 
  16.       this.maximumPoolSize = maximumPoolSize; 
  17.       this.workQueue = workQueue; 
  18.       this.keepAliveTime = unit.toNanos(keepAliveTime); 
  19.       this.threadFactory = threadFactory; 
  20.       this.handler = handler; 
  21.   } 

這里你可以看到最終賦值的過程,可以先大概知道下參數(shù)的意思:

corePoolSize:核心運(yùn)行的poolSize,也就是當(dāng)超過這個范圍的時候,就需要將新的Runnable放入到等待隊列workQueue中了,我們把這些Runnable就叫做要去執(zhí)行的任務(wù)吧。

maximumPoolSize:一般你用不到,當(dāng)大于了這個值就會將任務(wù)由一個丟棄處理機(jī)制來處理,但是當(dāng)你發(fā)生:newFixedThreadPool的時候,corePoolSizemaximumPoolSize是一樣的,而corePoolSize是先執(zhí)行的,所以他會先被放入等待隊列,而不會執(zhí)行到下面的丟棄處理中,看了后面的代碼你就知道了。

workQueue:等待隊列,當(dāng)達(dá)到corePoolSize的時候,就向該等待隊列放入線程信息(默認(rèn)為一個LinkedBlockingQueue),運(yùn)行中的線程屬性為:workers,為一個HashSet;我們的Runnable內(nèi)部被包裝了一層,后面會看到這部分代碼;這個隊列默認(rèn)是一個無界隊列(你也可以設(shè)定一個有界隊列),所以在生產(chǎn)者瘋狂生產(chǎn)的時候,考慮如何控制的問題。

keepAliveTime:默認(rèn)都是0,當(dāng)線程沒有任務(wù)處理后,保持多長時間,當(dāng)你使用:newCachedThreadPool(),它將是60s的時間。這個參數(shù)在運(yùn)行中的線程從workQueue獲取任務(wù)時,當(dāng)(poolSize >corePoolSize || allowCoreThreadTimeOut)會用到,當(dāng)然allowCoreThreadTimeOut要設(shè)置為true,也會先判定keepAliveTime是大于0的,不過由于它在corePoolSize上采用了Integer.MAX_VALUE,當(dāng)遇到系統(tǒng)遇到瞬間沖擊,workers就會迅速膨脹,所以這個地方就不要去設(shè)置allowCoreThreadTimeOut=true,否則結(jié)果是這些運(yùn)行中的線程會持續(xù)60s以上;另外,如果corePoolSize的值還沒到Integer.MAX_VALUE,當(dāng)超過那個值以后,這些運(yùn)行中的線程,也是

threadFactory:是構(gòu)造Thread的方法,你可以自己去包裝和傳遞,主要實現(xiàn)newThread方法即可;

handler:也就是參數(shù)maximumPoolSize達(dá)到后丟棄處理的方法,java提供了5種丟棄處理的方法,當(dāng)然你也可以自己根據(jù)實際情況去重寫,主要是要實現(xiàn)接口:RejectedExecutionHandler中的方法: public void rejectedExecution(Runnabler, ThreadPoolExecutor e) java默認(rèn)的是使用:AbortPolicy,他的作用是當(dāng)出現(xiàn)這中情況的時候會拋出一個異常;

其余的還包含:

  1. CallerRunsPolicy:如果發(fā)現(xiàn)線程池還在運(yùn)行,就直接運(yùn)行這個線程
  2. DiscardOldestPolicy:在線程池的等待隊列中,將頭取出一個拋棄,然后將當(dāng)前線程放進(jìn)去。
  3. DiscardPolicy:什么也不做
  4. AbortPolicy:java默認(rèn),拋出一個異常:RejectedExecutionException。

你可以自己寫一個,例如我們想在這個處理中,既不是完全丟棄,也不是完全啟動,也不是拋異常,而是控制生產(chǎn)者的線程,那么你就可以嘗試某種方式將生產(chǎn)者的線程blocking住,其實就有點類似提到的Semaphor的功能了。

通常你得到線程池后,會調(diào)用其中的:submit方法或execute方法 去操作;其實你會發(fā)現(xiàn),submit方法最終會調(diào)用execute方法來進(jìn)行操作,只是他提供了一個Future來托管返回值的處理而已,當(dāng)你調(diào)用需要有 返回值的信息時,你用它來處理是比較好的;這個Future會包裝對Callable信息,并定義一個Sync對象(),當(dāng)你發(fā)生讀取返回值的操作的時 候,會通過Sync對象進(jìn)入鎖,直到有返回值的數(shù)據(jù)通知,具體細(xì)節(jié)先不要看太多。

繼續(xù)向下,來看看execute最為核心的方法吧:

源碼段2:

  1. public void execute(Runnable command) { 
  2.     if (command == null
  3.         throw new NullPointerException(); 
  4.     if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) { 
  5.        if (runState == RUNNING && workQueue.offer(command)) { 
  6.            if (runState != RUNNING || poolSize == 0
  7.                ensureQueuedTaskHandled(command); 
  8.        } 
  9.        else if (!addIfUnderMaximumPoolSize(command)) 
  10.            reject(command); // is shutdown or saturated 
  11.    } 

這段代碼看似簡單,其實有點難懂,很多人也是這里沒看懂,沒事,我一個if一個if說:

首先第一個判定空操作就不用說了,下面判定的poolSize >= corePoolSize成立時候會進(jìn)入if的區(qū)域,當(dāng)然它不成立也有可能會進(jìn)入,他會判定addIfUnderCorePoolSize是否返回false,如果返回false就會進(jìn)去;

我們先來看下addIfUnderCorePoolSize方法的源碼是什么:

源碼段3:

  1. private boolean addIfUnderCorePoolSize(Runnable firstTask) { 
  2.     Thread t = null
  3.     final ReentrantLock mainLock = this.mainLock; 
  4.     mainLock.lock(); 
  5.     try { 
  6.         if (poolSize < corePoolSize && runState == RUNNING) 
  7.             t = addThread(firstTask); 
  8.     } finally { 
  9.         mainLock.unlock(); 
  10.     } 
  11.     if (t == null
  12.         return false
  13.     t.start(); 
  14.     return true

可以發(fā)現(xiàn),這段源碼是如果發(fā)現(xiàn)小雨corePoolSize就會創(chuàng)建一個新的線程,并且調(diào)用線程的start()方法將線程運(yùn)行起來:這個addThread()方法,我們先不考慮細(xì)節(jié),因為我們還要先看到前面是怎么進(jìn)去的,這里可以發(fā)信啊,只有沒有創(chuàng)建成功Thread才會返回false,也就是當(dāng)當(dāng)前的poolSize > corePoolSize的時候,或線程池已經(jīng)不是在running狀態(tài)的時候才會出現(xiàn);

注意:這里在外部判定一次poolSize和corePoolSize只是初步判定,內(nèi)部是加鎖后判定的,以得到更為準(zhǔn)確的結(jié)果,而外部初步判定如果是大于了,就沒有必要進(jìn)入這段有鎖的代碼了。

此時我們知道了,當(dāng)前線程數(shù)量大于corePoolSize的時候,就會進(jìn)入【代碼段2】的第一個if語句中,回到【源碼段2】,繼續(xù)看if語句中的內(nèi)容:

這里標(biāo)記為

源碼段4:

  1. if (runState == RUNNING && workQueue.offer(command)) { 
  2.     if (runState != RUNNING || poolSize == 0
  3.         ensureQueuedTaskHandled(command); 
  4. else if (!addIfUnderMaximumPoolSize(command)) 
  5.     reject(command); // is shutdown or saturated 

第一個if,也就是當(dāng)當(dāng)前狀態(tài)為running的時候,就會去執(zhí)行workQueue.offer(command),這個workQueue其實 就是一個BlockingQueue,offer()操作就是在隊列的尾部寫入一個對象,此時寫入的對象為線程的對象而已;所以你可以認(rèn)為只有線程池在 RUNNING狀態(tài),才會在隊列尾部插入數(shù)據(jù),否則就執(zhí)行else if,其實else if可以看出是要做一個是否大于MaximumPoolSize的判定,如果大于這個值,就會做reject的操作,關(guān)于reject的說明,我們在【源 碼段1】的解釋中已經(jīng)非常明確的說明,這里可以簡單看下源碼,以應(yīng)征結(jié)果:

源碼段5:

  1. private boolean addIfUnderMaximumPoolSize(Runnable firstTask) { 
  2.     Thread t = null
  3.     final ReentrantLock mainLock = this.mainLock; 
  4.     mainLock.lock(); 
  5.     try { 
  6.         if (poolSize < maximumPoolSize && runState == RUNNING)                 
  7. //在corePoolSize = maximumPoolSize下,該代碼幾乎不可能運(yùn)行                 
  8. t = addThread(firstTask);       
  9.    } finally {            
  10.   mainLock.unlock();         }        
  11.   if (t == null)           
  12.    return false;         
  13. t.start();        
  14.  return true; } 
  15. void reject(Runnable command) {         
  16. handler.rejectedExecution(command, this);     }  

也就是如果線程池滿了,而且線程池調(diào)用了shutdown后,還在調(diào)用execute方法時,就會拋出上面說明的異常:RejectedExecutionException 再回頭來看下【代碼段4】中進(jìn)入到等待隊列后的操作:

  1. if (runState != RUNNING || poolSize == 0)     ensureQueuedTaskHandled(command);  

這段代碼是要在線程池運(yùn)行狀態(tài)不是RUNNING或poolSize == 0才會調(diào)用,他是干啥呢? 他為什么會不等于RUNNING呢?外面那一層不是判定了他== RUNNING了么,其實有時間差就是了,如果是poolSize == 0也會執(zhí)行這段代碼,但是里面的判定條件是如果不是RUNNING,就做reject操作,在第一個線程進(jìn)去的時候,會將第一個線程直接啟動起來;很多人 也是看這段代碼很繞,因為不斷的循環(huán)判定類似的判定條件,你主要記住他們之間有時間差,要取最新的就好了。 此時貌似代碼看完了?咦,此時有問題了: 1、 等待中的線程在后來是如何跑起來的呢?線程池是不是有類似Timer一樣的守護(hù)進(jìn)程不斷掃描線程隊列和等待隊列?還是利用某種鎖機(jī)制,實現(xiàn)類似wait和 notify實現(xiàn)的? 2、 線程池的運(yùn)行隊列和等待隊列是如何管理的呢?這里還沒看出影子呢! NO,NO,NO! Java在實現(xiàn)這部分的時候,使用了怪異的手段,神馬手段呢,還要再看一部分代碼才曉得。 在前面【源碼段3】中,我們看到了一個方法叫:addThread(),也許很少有人會想到關(guān)鍵在這里,其實關(guān)鍵就是在這里: 我們看看addThread()方法到底做了什么。

源碼段6:

  1.  private Thread addThread(Runnable firstTask) {         
  2. Worker w = new Worker(firstTask);         
  3. Thread t = threadFactory.newThread(w);        
  4.  if (t != null) {             
  5. w.thread = t;             
  6. workers.add(w);             
  7. int nt = ++poolSize;             
  8. if (nt > largestPoolSize) 
  9.             largestPoolSize = nt; 
  10.     } 
  11.     return t; 

這里創(chuàng)建了一個Worker,其余的操作,就是將poolSize++的操作,然后將將其放入workers的運(yùn)行的HashSet中等操作;

我們主要關(guān)心Worker是干什么的,因為這個threadFactory對 我們用途不大,只是做了Thread的命名處理;而Worker你會發(fā)現(xiàn)它的定義也是一個Runnable,外部開始在代碼段中發(fā)現(xiàn)了調(diào)用哪個這個 Worker的start()方法,也就是線程的啟動方法,其實也就是調(diào)用了Worker的run()方法,那么我們重點要關(guān)心run方法是如何處理的

源碼段7:

  1. public void run() { 
  2.      try { 
  3.          Runnable task = firstTask; 
  4.          firstTask = null
  5.          while (task != null || (task = getTask()) != null) { 
  6.              runTask(task); 
  7.              task = null
  8.          } 
  9.      } finally { 
  10.          workerDone(this); 
  11.      } 
  12.  } 

FirstTask其實就是開始在創(chuàng)建work的時候,由外部傳入的Runnable對象,也就是你自己的Thread,你會發(fā)現(xiàn)它如果發(fā)現(xiàn)task為空,就會調(diào)用getTask()方法再判定,直到兩者為空,并且是一個while循環(huán)體。

那么看看getTask()方法的實現(xiàn)為:

源碼段8:

  1. Runnable getTask() { 
  2.    for (;;) { 
  3.        try { 
  4.            int state = runState; 
  5.            if (state > SHUTDOWN) 
  6.                return null
  7.            Runnable r; 
  8.            if (state == SHUTDOWN)  // Help drain queue 
  9.                r = workQueue.poll(); 
  10.            else if (poolSize > corePoolSize || allowCoreThreadTimeOut) 
  11.                r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS); 
  12.            else 
  13.                r = workQueue.take(); 
  14.            if (r != null
  15.                return r; 
  16.            if (workerCanExit()) { 
  17.                if (runState >= SHUTDOWN) // Wake up others 
  18.                    interruptIdleWorkers(); 
  19.                return null
  20.            } 
  21.            // Else retry 
  22.        } catch (InterruptedException ie) { 
  23.            // On interruption, re-check runState 
  24.        } 
  25.    } 

你會發(fā)現(xiàn)它是從workQueue隊列中,也就是等待隊列中獲取一個元素出來并返回!

回過頭來根據(jù)代碼段6理解下:

當(dāng)前線程運(yùn)行完后,在到workQueue中去獲取一個task出來,繼續(xù)運(yùn)行,這樣就保證了線程池中有一定的線程一直在運(yùn)行;此時若跳出了 while循環(huán),只有workQueue隊列為空才會出現(xiàn)或出現(xiàn)了類似于shutdown的操作,自然運(yùn)行隊列會減少1,當(dāng)再有新的線程進(jìn)來的時候,就又 開始向worker里面放數(shù)據(jù)了,這樣以此類推,實現(xiàn)了線程池的功能。

這里可以看下run方法的finally中調(diào)用的workerDone方法為:

源碼段9:

  1. void workerDone(Worker w) { 
  2.     final ReentrantLock mainLock = this.mainLock; 
  3.     mainLock.lock(); 
  4.     try { 
  5.         completedTaskCount += w.completedTasks; 
  6.         workers.remove(w); 
  7.         if (--poolSize == 0
  8.             tryTerminate(); 
  9.     } finally { 
  10.         mainLock.unlock(); 
  11.     } 

注意這里將workers.remove(w)掉,并且調(diào)用了—poolSize來做操作。

至于tryTerminate是做了更多關(guān)于回收方面的操作。

最后我們還要看一段代碼就是在【源碼段6】中出現(xiàn)的代碼調(diào)用為:runTask(task);這個方法也是運(yùn)行的關(guān)鍵。

源碼段10:

  1. private void runTask(Runnable task) { 
  2.        final ReentrantLock runLock = this.runLock; 
  3.        runLock.lock(); 
  4.        try { 
  5.            if (runState < STOP &&                     Thread.interrupted() &&                     runState >= STOP) 
  6.                thread.interrupt(); 
  7.  
  8.            boolean ran = false
  9.            beforeExecute(thread, task); 
  10.            try { 
  11.                task.run(); 
  12.                ran = true
  13.                afterExecute(task, null); 
  14.                ++completedTasks; 
  15.            } catch (RuntimeException ex) { 
  16.                if (!ran) 
  17.                    afterExecute(task, ex); 
  18.                throw ex; 
  19.            } 
  20.        } finally { 
  21.            runLock.unlock(); 
  22.        } 
  23.    } 

你可以看到,這里面的task為傳入的task信息,調(diào)用的不是start方法,而是run方法,因為run方法直接調(diào)用不會啟動新的線程,也是因為這樣,導(dǎo)致了你無法獲取到你自己的線程的狀態(tài),因為線程池是直接調(diào)用的run方法,而不是start方法來運(yùn)行。

這里有個beforeExecuteafterExecute方法,分別代表在執(zhí)行前和執(zhí)行后,你可以做一段操作,在這個類中,這兩個方法都是【空body】的,因為普通線程池?zé)o需做更多的操作。

如果你要實現(xiàn)類似暫停等待通知的或其他的操作,可以自己extends后進(jìn)行重寫構(gòu)造;

本文沒有介紹關(guān)于ScheduledThreadPoolExecutor調(diào)用的細(xì)節(jié),下一篇文章會詳細(xì)說明,因為大部分代碼和本文一致,區(qū)別在于一些細(xì)節(jié),在介紹:ScheduledThreadPoolExecutor的時候,會明確的介紹它與Timer和TimerTask的巨大區(qū)別,區(qū)別不在于使用,而是在于本身內(nèi)在的處理細(xì)節(jié)。

責(zé)任編輯:陳四芳 來源: ifeve.com
相關(guān)推薦

2015-10-10 09:39:42

Java線程池源碼解析

2021-05-26 11:30:24

Java線程池代碼

2022-12-16 08:31:37

調(diào)度線程池源碼

2025-09-24 18:39:45

2018-10-31 15:54:47

Java線程池源碼

2011-08-19 17:36:42

iPhone操作隊列Java

2020-11-25 11:33:47

Java線程技術(shù)

2020-12-08 08:53:53

編程ThreadPoolE線程池

2012-05-15 02:18:31

Java線程池

2013-05-28 13:57:12

MariaDB

2011-06-22 15:50:45

QT 線程

2024-11-27 08:15:50

2013-06-08 13:07:23

Java線程池調(diào)度器

2021-07-16 11:35:20

Java線程池代碼

2023-11-29 16:38:12

線程池阻塞隊列開發(fā)

2020-12-10 08:24:40

線程池線程方法

2021-09-11 07:32:15

Java線程線程池

2021-09-11 15:26:23

Java多線程線程池

2020-12-10 07:00:38

編程線程池定時任務(wù)

2020-10-10 08:20:27

Spring Boot運(yùn)行原理代碼
點贊
收藏

51CTO技術(shù)棧公眾號