av激情亚洲男人的天堂国语,日韩欧美精品一中文字幕,无码av一区二区三区无码,国产又色又爽又刺激的a片,国产又色又爽又刺激的a片

Java并發(fā)：jucExecutor框架詳解

Executor 框架是 juc 里提供的線程池的實現(xiàn)。前兩天看了下 Executor 框架的一些源碼，做個簡單的總結。

10年積累的成都做網(wǎng)站、網(wǎng)站設計經(jīng)驗，可以快速應對客戶對網(wǎng)站的新想法和需求。提供各種問題對應的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡服務。我雖然不認識你，你也不認識我。但先網(wǎng)站設計后付款的網(wǎng)站建設流程，更有山東免費網(wǎng)站建設讓你可以放心的選擇與我們合作。

線程池大概的思路是維護一個的線程池用于執(zhí)行提交的任務。我理解池的技術的主要意義有兩個：

1. 資源的控制，如并發(fā)量限制。像連接池這種是對數(shù)據(jù)庫資源的保護。

2. 資源的有效利用，如線程復用，避免頻繁創(chuàng)建線程和線程上下文切換。

那么想象中設計一個線程池就需要有線程池大小、線程生命周期管理、等待隊列等等功能，下面結合代碼看看原理。

Excutor 整體結構如下：

Executor 接口定義了最基本的 execute 方法，用于接收用戶提交任務。 ExecutorService 定義了線程池終止和創(chuàng)建及提交 futureTask 任務支持的方法。

AbstractExecutorService 是抽象類，主要實現(xiàn)了 ExecutorService 和 futureTask 相關的一些任務創(chuàng)建和提交的方法。

ThreadPoolExecutor 是最核心的一個類，是線程池的內(nèi)部實現(xiàn)。線程池的功能都在這里實現(xiàn)了，平時用的最多的基本就是這個了。其源碼很精練，遠沒當時想象的多。

ScheduledThreadPoolExecutor 在 ThreadPoolExecutor 的基礎上提供了支持定時調(diào)度的功能。線程任務可以在一定延時時間后才被觸發(fā)執(zhí)行。

1、ThreadPoolExecutor 原理

1.1 ThreadPoolExecutor內(nèi)部的幾個重要屬性

1.線程池本身的狀態(tài)

 
 
 
 
    
  
  
  
  
  1. volatile int runState;  
    2. 
  
  
  
  
  2. static final int RUNNING = 0;  
    3. 
  
  
  
  
  3. static final int SHUTDOWN = 1;  
    4. 
  
  
  
  
  4. static final int STOP = 2;  
    5. 
  
  
  
  
  5. static final int TERMINATED = 3; 
    6.

2.等待任務隊列和工作集

 
 
 
 
    
  
  
  
  
  1. private final BlockingQueue workQueue; //等待被執(zhí)行的Runnable任務  
    2. 
  
  
  
  
  2. private final HashSet workers = new HashSet(); //正在被執(zhí)行的Worker任務集 
    3.

 3.線程池的主要狀態(tài)鎖。線程池內(nèi)部的狀態(tài)變化 ( 如線程大小 ) 都需要基于此鎖。

 
 
 
 
    
  
  
  
  
  1. private final ReentrantLock mainLock = new ReentrantLock();
    2.

4.線程的存活時間和大小

 
 
 
 
    
  
  
  
  
  1. private volatile long keepAliveTime;// 線程存活時間  
    2. 
  
  
  
  
  2. private volatile boolean allowCoreThreadTimeOut;// 是否允許核心線程存活  
    3. 
  
  
  
  
  3. private volatile int corePoolSize;// 核心池大小  
    4. 
  
  
  
  
  4. private volatile int maximumPoolSize; // 最大池大小  
    5. 
  
  
  
  
  5. private volatile int poolSize; //當前池大小  
    6. 
  
  
  
  
  6. private int largestPoolSize; //最大池大小，區(qū)別于maximumPoolSize，是用于記錄線程池曾經(jīng)達到過的最大并發(fā),理論上小于等于maximumPoolSize。  
    7.

5.線程工廠和拒絕策略

 
 
 
 
    
  
  
  
  
  1. private volatile RejectedExecutionHandler handler;// 拒絕策略，用于當線程池無法承載新線程是的處理策略。 
    2. 
  
  
  
  
  2.  private volatile ThreadFactory threadFactory;// 線程工廠，用于在線程池需要新創(chuàng)建線程的時候創(chuàng)建線程
    3.

6.線程池完成任務數(shù)

 
 
 
 
    
  
  
  
  
  1. private long completedTaskCount;//線程池運行到當前完成的任務數(shù)總和 
    2.

1.2 ThreadPoolExecutor 的內(nèi)部工作原理

有了以上定義好的數(shù)據(jù)，下面來看看內(nèi)部是如何實現(xiàn)的 。 Doug Lea 的整個思路總結起來就是 5 句話：

  1. 如果當前池大小 poolSize 小于 corePoolSize ，則創(chuàng)建新線程執(zhí)行任務。
  2. 如果當前池大小 poolSize 大于 corePoolSize ，且等待隊列未滿，則進入等待隊列
  3. 如果當前池大小 poolSize 大于 corePoolSize 且小于 maximumPoolSize ，且等待隊列已滿，則創(chuàng)建新線程執(zhí)行任務。
  4. 如果當前池大小 poolSize 大于 corePoolSize 且大于 maximumPoolSize ，且等待隊列已滿，則調(diào)用拒絕策略來處理該任務。
  5. 線程池里的每個線程執(zhí)行完任務后不會立刻退出，而是會去檢查下等待隊列里是否還有線程任務需要執(zhí)行，如果在 keepAliveTime 里等不到新的任務了，那么線程就會退出。

下面看看代碼實現(xiàn)：

線程池最重要的方法是由 Executor 接口定義的 execute 方法 , 是任務提交的入口。

我們看看 ThreadPoolExecutor.execute(Runnable cmd) 的實現(xiàn)：

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

解釋如下：

當提交一個新的 Runnable 任務：

分支1 ： 如果當前池大小小于 corePoolSize, 執(zhí)行 addIfUnderCorePoolSize(command) , 如果線程池處于運行狀態(tài)且 poolSize < corePoolSize addIfUnderCorePoolSize(command) 會做如下事情，將 Runnable 任務封裝成 Worker 任務 , 創(chuàng)建新的 Thread ，執(zhí)行 Worker 任務。如果不滿足條件，則返回 false 。

代碼如下：

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

分支2 ： 如果大于 corePoolSize 或 1 失敗失敗，則：

  • 如果等待隊列未滿，把 Runnable 任務加入到 workQueue 等待隊列
    workQueue .offer(command)
  • 如多等待隊列已經(jīng)滿了，調(diào)用 addIfUnderMaximumPoolSize(command) ，和 addIfUnderCorePoolSize 基本類似，只不過判斷條件是 poolSize < maximumPoolSize 。如果大于 maximumPoolSize ，則把 Runnable 任務交由 RejectedExecutionHandler 來處理。

問題：如何實現(xiàn)線程的復用？

Doug Lea 的實現(xiàn)思路是 線程池里的每個線程執(zhí)行完任務后不立刻退出，而是去檢查下等待隊列里是否還有線程任務需要執(zhí)行，如果在 keepAliveTime 里等不到新的任務了，那么線程就會退出。這個功能的實現(xiàn) 關鍵在于 Worker 。線程池在執(zhí)行 Runnable 任務的時候，并不單純把 Runnable 任務交給創(chuàng)建一個 Thread 。而是會把 Runnable 任務封裝成 Worker 任務。

下面看看 Worker 的實現(xiàn)：

代碼很簡單，可以看出， worker 里面包裝了 firstTask 屬性，在構造worker 的時候傳進來的那個 Runnable 任務就是 firstTask 。 同時也實現(xiàn)了Runnable接口，所以是個代理模式，看看代理增加了哪些功能。 關鍵看 woker 的 run 方法：

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

可以看出 worker 的 run 方法是一個循環(huán)，第一次循環(huán)運行的必然是 firstTask ，在運行完 firstTask 之后，并不會立刻結束，而是會調(diào)用 getTask 獲取新的任務（ getTask 會從等待隊列里獲取等待中的任務），如果 keepAliveTime 時間內(nèi)得到新任務則繼續(xù)執(zhí)行，得不到新任務則那么線程才會退出。這樣就保證了多個任務可以復用一個線程，而不是每次都創(chuàng)建新任務。 keepAliveTime 的邏輯在哪里實現(xiàn)的呢？主要是利用了 BlockingQueue 的 poll 方法支持等待?？煽?getTask 的代碼段：

 
 
 
 
    
  
  
  
  
  1. if (state == SHUTDOWN)  // Help drain queue 
    2. 
  
  
  
  
  2.     r = workQueue.poll(); 
    3. 
  
  
  
  
  3. else if (poolSize > corePoolSize || allowCoreThreadTimeOut) 
    4. 
  
  
  
  
  4.     r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS); 
    5. 
  
  
  
  
  5. else
    6. 
  
  
  
  
  6.     r = workQueue.take(); 
    7.

2.ThreadFactory 和R ejectedExecutionHandler

ThreadFactory 和 RejectedExecutionHandler是ThreadPoolExecutor的兩個屬性，也 可以認為是兩個簡單的擴展點 . ThreadFactory 是創(chuàng)建線程的工廠。

默認的線程工廠會創(chuàng)建一個帶有“ pool-poolNumber-thread-threadNumber ”為名字的線程，如果我們有特別的需要，如線程組命名、優(yōu)先級等，可以定制自己的 ThreadFactory 。

RejectedExecutionHandler 是拒絕的策略。常見有以下幾種：

AbortPolicy ：不執(zhí)行，會拋出 RejectedExecutionException 異常。

CallerRunsPolicy ：由調(diào)用者（調(diào)用線程池的主線程）執(zhí)行。

DiscardOldestPolicy ：拋棄等待隊列中最老的。

DiscardPolicy: 不做任何處理，即拋棄當前任務。

3.ScheduledThreadPoolExecutor

ScheduleThreadPoolExecutor 是對ThreadPoolExecutor的集成。增加了定時觸發(fā)線程任務的功能。需要注意

從內(nèi)部實現(xiàn)看， ScheduleThreadPoolExecutor 使用的是 corePoolSize 線程和一個無界隊列的固定大小的池，所以調(diào)整 maximumPoolSize 沒有效果。無界隊列是一個內(nèi)部自定義的 DelayedWorkQueue 。

ScheduleThreadPoolExecutor 線程池接收定時任務的方法是 schedule ，看看內(nèi)部實現(xiàn)：

 
 
 
 
    
  
  
  
  
  1. public ScheduledFuture schedule(Runnable command, 
    2. 
  
  
  
  
  2.                                    long delay, 
    3. 
  
  
  
  
  3.                                    TimeUnit unit) { 
    4. 
  
  
  
  
  4.     if (command == null || unit == null) 
    5. 
  
  
  
  
  5.         throw new NullPointerException(); 
    6. 
  
  
  
  
  6.     RunnableScheduledFuture t = decorateTask(command, 
    7. 
  
  
  
  
  7.         new ScheduledFutureTask(command, null, 
    8. 
  
  
  
  
  8.                                       triggerTime(delay, unit))); 
    9. 
  
  
  
  
  9. 
  
  
  
  
  10.     delayedExecute(t); 
    11. 
  
  
  
  
  11.     return t; 
    12. 
  
  
  
  
  12. } 
    13.

以上代碼會初始化一個 RunnableScheduledFuture 類型的任務 t, 并交給 delayedExecute 方法。 delayedExecute(t) 方法實現(xiàn)如下：

 
 
 
 
    
  
  
  
  
  1.     private void delayedExecute(Runnable command) { 
    2. 
  
  
  
  
  2.         if (isShutdown()) { 
    3. 
  
  
  
  
  3.             reject(command); 
    4. 
  
  
  
  
  4.             return; 
    5. 
  
  
  
  
  5.         } 
    6. 
  
  
  
  
  6.         if (getPoolSize() < getCorePoolSize()) 
    7. 
  
  
  
  
  7.             prestartCoreThread(); 
    8. 
  
  
  
  
  8. 
  
  
  
  
  9.         super.getQueue().add(command); 
    10. 
  
  
  
  
  10. } 
    11.

如果當前線程池大小 poolSize 小于 CorePoolSize ，則創(chuàng)建一個新的線程，注意這里創(chuàng)建的線程是空的，不會把任務直接交給線程來做，而是把線程任務放到隊列里。因為任務是要定時觸發(fā)的，所以不能直接交給線程去執(zhí)行。

問題： 那如何做到定時觸發(fā)呢？

關鍵在于DelayedWorkQueue,它代理了 DelayQueue ?？梢哉J為 DelayQueue 是這樣一個隊列（具體可以去看下源碼，不詳細分析）：

1. 隊列里的元素按照任務的 delay 時間長短升序排序， delay 時間短的在隊頭， delay 時間長的在隊尾。

2. 從 DelayQueue 里 FIFO 的獲取一個元素的時候，不會直接返回 head 。可能會阻塞，等到 head 節(jié)點到達 delay 時間后才能被獲取?？梢钥聪?DelayQueue 的 take 方法實現(xiàn)：

 
 
 
 
    
  
  
  
  
  1. public E take() throws InterruptedException { 
    2. 
  
  
  
  
  2.     final ReentrantLock lock = this.lock; 
    3. 
  
  
  
  
  3.     lock.lockInterruptibly(); 
    4. 
  
  
  
  
  4.     try { 
    5. 
  
  
  
  
  5.         for (;;) { 
    6. 
  
  
  
  
  6.             E first = q.peek(); 
    7. 
  
  
  
  
  7.             if (first == null) { 
    8. 
  
  
  
  
  8.                 available.await(); 
    9. 
  
  
  
  
  9.             } else { 
    10. 
  
  
  
  
  10.                 long delay =  first.getDelay(TimeUnit.NANOSECONDS); 
    11. 
  
  
  
  
  11.                 if (delay > 0) { 
    12. 
  
  
  
  
  12.                     long tl = available.awaitNanos(delay);//等待delay時間 
    13. 
  
  
  
  
  13.                 } else { 
    14. 
  
  
  
  
  14.                     E x = q.poll(); 
    15. 
  
  
  
  
  15.                     assert x != null; 
    16. 
  
  
  
  
  16.                     if (q.size() != 0) 
    17. 
  
  
  
  
  17.                         available.signalAll(); // wake up other takers 
    18. 
  
  
  
  
  18.                     return x; 
    19. 
  
  
  
  
  19.                 } 
    20. 
  
  
  
  
  20.             } 
    21. 
  
  
  
  
  21.         } 
    22. 
  
  
  
  
  22.     } finally { 
    23. 
  
  
  
  
  23.         lock.unlock(); 
    24. 
  
  
  
  
  24.     } 
    25. 
  
  
  
  
  25. }
    26.

4.線程池使用策略

通過以上的詳解基本上能夠定制出自己需要的策略了，下面簡單介紹下Executors里面提供的一些常見線程池策略：

1.FixedThreadPool

 
 
 
 
    
  
  
  
  
  1. public static ExecutorService newFixedThreadPool(int nThreads) { 
    2. 
  
  
  
  
  2.     return new ThreadPoolExecutor(nThreads, nThreads, 
    3. 
  
  
  
  
  3.                                   0L, TimeUnit.MILLISECONDS, 
    4. 
  
  
  
  
  4.                                   new LinkedBlockingQueue()); 
    5. 
  
  
  
  
  5. }
    6.

實際上就是個不支持keepalivetime，且corePoolSize和maximumPoolSize相等的線程池。

2.SingleThreadExecutor

 
 
 
 
    
  
  
  
  
  1. public static ExecutorService newSingleThreadExecutor() { 
    2. 
  
  
  
  
  2.     return new FinalizableDelegatedExecutorService 
    3. 
  
  
  
  
  3.         (new ThreadPoolExecutor(1, 1, 
    4. 
  
  
  
  
  4.                                 0L, TimeUnit.MILLISECONDS, 
    5. 
  
  
  
  
  5.                                 new LinkedBlockingQueue())); 
    6. 
  
  
  
  
  6. }
    7.

實際上就是個不支持keepalivetime，且corePoolSize和maximumPoolSize都等1的線程池。

3.CachedThreadPool

 
 
 
 
    
  
  
  
  
  1. public static ExecutorService newCachedThreadPool() { 
    2. 
  
  
  
  
  2.     return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 
    3. 
  
  
  
  
  3.                                   60L, TimeUnit.SECONDS, 
    4. 
  
  
  
  
  4.                                   new SynchronousQueue()); 
    5. 
  
  
  
  
  5. }
    6.

實際上就是個支持keepalivetime時間是60秒（線程空閑存活時間），且corePoolSize為0，maximumPoolSize無窮大的線程池。

4.SingleThreadScheduledExecutor

 
 
 
 
    
  
  
  
  
  1. public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) { 
    2. 
  
  
  
  
  2.     return new DelegatedScheduledExecutorService 
    3. 
  
  
  
  
  3.         (new ScheduledThreadPoolExecutor(1, threadFactory)); 
    4. 
  
  
  
  
  4. } 
    5.

實際上是個corePoolSize為1的ScheduledExecutor。上文說過ScheduledExecutor采用無界等待隊列，所以maximumPoolSize沒有作用。

5.ScheduledThreadPool

 
 
 
 
    
  
  
  
  
  1. public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { 
    2. 
  
  
  
  
  2.     return new ScheduledThreadPoolExecutor(corePoolSize); 
    3. 
  
  
  
  
  3. }
    4.

實際上是corePoolSize課設定的ScheduledExecutor。上文說過ScheduledExecutor采用無界等待隊列，所以maximumPoolSize沒有作用。

以上還不一定滿足你的需要，完全可以根據(jù)自己需要去定制。


網(wǎng)站標題：Java并發(fā)：jucExecutor框架詳解
網(wǎng)站鏈接：http://uogjgqi.cn/article/djiochj.html
掃二維碼與項目經(jīng)理溝通

我們在微信上24小時期待你的聲音

解答本文疑問/技術咨詢/運營咨詢/技術建議/互聯(lián)網(wǎng)交流