多线程池、饱和策略详解
一、序言
當我們需要使用線程的時候，我們可以新建一個線程，然后顯式調用線程的start()方法，這樣實現起來非常簡便，但在某些場景下存在缺陷：如果需要同時執行多個任務（即并發的線程數量很多），頻繁地創建線程會降低系統的效率，因為創建和銷毀線程均需要一定的時間。
線程池可以使線程得到復用，所謂線程復用就是線程在執行完一個任務后并不被銷毀，該線程可以繼續執行其他的任務。java.lang.concurrent包中的Executors類為我們創建線程池提供了方便。
二、Executors的簡單使用示例
此處我們先來看一個簡單的例子，如下：
package com.soft; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class ExecutorsDemo { public static void main(String[] args) throws InterruptedException, ExecutionException { // ExecutorService executor = Executors.newSingleThreadExecutor(); // ExecutorService executor = Executors.newCachedThreadPool();ExecutorService executor = Executors.newFixedThreadPool(5);Thread.sleep(5*1000);//方便監控工具能捕獲到for (int i = 0; i < 10; i++) { final int no = i;Runnable runnable = new Runnable() { public void run() { try {System.out.println("into" + no);Thread.sleep(1000L);System.out.println("end" + no);} catch (InterruptedException e) {e.printStackTrace();}}};executor.execute(runnable);//ExecutorService有一個execute()方法，這個方法的參數是Runnable類型，通過execute(Runnable)方法即可將一個任務添加到線程池，任務的執行方法是Runnable類型對象的run()方法。}//End forexecutor.shutdown();System.out.println("Thread Main End!");} }復制代碼其運行結果如下：
into0 into3 Thread Main End! into4 into1 into2 end0 into5 end3 end1 end4 into8 into6 into7 end2 into9 end5 end7 end8 end6 end9復制代碼解說：這個例子應該很容易看懂，從運行結果來看，在任意某一時刻只有5個線程在執行，這是因為上述代碼通過Executors.newFixedThreadPool(5)語句創建了一個固定長度的線程池（長度為5），一個結束之后另再一個才開始執行。
三、Executors提供的線程池
Executors是線程的工廠類，也可以說是一個線程池工具類，它調用其內部靜態方法（如newFixedThreadPool()等）即可創建一個線程池，通過參數設置，Executors提供不同的線程池機制。
四、簡述線程池的屬性
五、詳解ThreadPoolExecutor##
上文提到可以通過顯式的ThreadPoolExecutor構造函數來構造特定形式的線程池，ThreadPoolExecutor是java.util.concurrent包以內部線程池的形式對外提供線程池管理、線程調度等服務，此處我們來了解一下ThreadPoolExecutor
- （1）一般使用方式：
下文詳解此示例涉及的一些內容
- 2）構造函數的聲明：
- （3）函數參數說明：
| corePoolSize | 線程池的基本大小（核心線程池大小） |
| maximumPoolSize | 線程池的最大大小 |
| keepAliveTime | 線程池中超過corePoolSize數目的空閑線程的最大存活時間 |
| unit | keepAliveTime參數的時間單位 |
| workQueue | 任務阻塞隊列 |
| threadFactory | 新建線程的工廠 |
| handler | 當提交的任務數超過maxmumPoolSize與workQueue之和時，任務會交給RejectedExecutionHandler來處理 |
進一步解說：
-
A、當提交新任務時，若線程池大小小于corePoolSize，將創建一個新的線程來執行任務，即使此時線程池中存在空閑線程；
-
B、當提交新任務時，若線程池達到corePoolSize大小，新提交的任務將被放入workQueue中，等待線程池調度執行；
-
C、當提交新任務時，若workQueue已滿，且maximumPoolSize>corePoolSize，將創建新的線程來執行任務；
-
D、當提交新任務時，若任務總數超過maximumPoolSize，新提交的任務將由RejectedExecutionHandler來處理；
-
E、當線程池中的線程數超過corePoolSize時，若線程的空閑時間達到keepAliveTime，則關閉空閑線程
-
（4）任務阻塞隊列選擇機制
- （5）簡述SynchronousQueue
注：此處貼出SynchronousQueue的使用示例，示例中使用了Semaphore，更多關于SynchronousQueue及Semaphore的內容請參考其他文章
package com.test; import java.util.concurrent.Semaphore; import java.util.concurrent.SynchronousQueue; /* * 程序中有10個線程來消費生成者產生的數據，這些消費者都調用TestDo.doSome()方法去進行處理，* 每個消費者都需要一秒才能處理完，程序應保證這些消費者線程依次有序地消費數據，只有上一個消費者消費完后，* 下一個消費者才能消費數據，下一個消費者是誰都可以，但要保證這些消費者線程拿到的數據是有順序的。 */ public class SynchronousQueueTest { public static void main(String[] args) {System.out.println("begin:" + (System.currentTimeMillis() / 1000)); // 定義一個Synchronousfinal SynchronousQueue<String> sq = new SynchronousQueue<String>(); // 定義一個數量為1的信號量，其作用相當于一個互斥鎖final Semaphore sem = new Semaphore(1); for (int i = 0; i < 10; i++) { new Thread(new Runnable() { public void run() { try {sem.acquire();String input = sq.take();String output = TestDo.doSome(input);//內部類System.out.println(Thread.currentThread().getName()+ ":" + output);sem.release();} catch (InterruptedException e) {e.printStackTrace();}}}).start();} for (int i = 0; i < 10; i++) {String input = i + ""; //此處將i變成字符串try {sq.put(input);} catch (InterruptedException e) {e.printStackTrace();}}}//End main} class TestDo { public static String doSome(String input) { try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}String output = input + ":" + (System.currentTimeMillis() / 1000); return output;} }復制代碼上述代碼的運行結果如下：
begin:1458954798 Thread-0:0:1458954799 Thread-1:1:1458954800 Thread-2:2:1458954801 Thread-3:3:1458954802 Thread-4:4:1458954803 Thread-5:5:1458954804 Thread-6:6:1458954805 Thread-7:7:1458954806 Thread-8:8:1458954807 Thread-9:9:1458954808復制代碼從上述結果看，上例在任意某一時刻只有一個線程在執行，且只有前一個線程執行完下一個線程才開始
六、飽和策略（線程池任務拒絕策略）
上文提到ThreadPoolExecutor構造函數的RejectedExecutionHandler handler參數，該參數表示當提交的任務數超過maxmumPoolSize與workQueue之和時，任務會交給RejectedExecutionHandler來處理，此處我們來具體了解一下
*（1）四種飽和策略
*（2）源碼分析：
RejectedExecutionHandler這個接口是用來處理被丟棄的線程的異常處理接口，其源碼如下：
public interface RejectedExecutionHandler{ //被線程池丟棄的線程處理機制public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) ; }復制代碼- AbortPolicy（中止策略）繼承RejectedExecutionHandler接口，其源碼如下：
我們可以自己實現RejectedExecutionHandler接口，將實現類作為線程丟棄處理類，代碼如下：
package com.test; import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadPoolExecutor; public class RejectedExecutionHandlerDemo implements RejectedExecutionHandler{@Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // TODO Auto-generated method stubSystem.out.println("線程信息"+r.toString()+"被遺棄的線程池："+executor.toString());}}復制代碼七、定制ThreadPoolExecutor
*（1）通過修改參數的方式達到定制目的
*（2）通過自定義方式（封裝各種參數）達到定制目的
import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class CustomThreadPoolExecutor { private ThreadPoolExecutor pool = null; /** * 線程池初始化方法 * * corePoolSize 核心線程池大小----10 * maximumPoolSize 最大線程池大小----30 * keepAliveTime 線程池中超過corePoolSize數目的空閑線程最大存活時間----30+單位TimeUnit * TimeUnit keepAliveTime時間單位----TimeUnit.MINUTES * workQueue 阻塞隊列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞隊列 * threadFactory 新建線程工廠----new CustomThreadFactory()====定制的線程工廠 * rejectedExecutionHandler 當提交任務數超過maxmumPoolSize+workQueue之和時, * 即當提交第41個任務時(前面線程都沒有執行完,此測試方法中用sleep(100)), * 任務會交給RejectedExecutionHandler來處理 */ public void init() { pool = new ThreadPoolExecutor( 10, 30, 30, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(10), new CustomThreadFactory(),new CustomRejectedExecutionHandler()); } public void destory() { if(pool != null) { pool.shutdownNow(); } } public ExecutorService getCustomThreadPoolExecutor() { return this.pool; } private class CustomThreadFactory implements ThreadFactory { private AtomicInteger count = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { Thread t = new Thread(r); String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1); System.out.println(threadName); t.setName(threadName); return t; } } private class CustomRejectedExecutionHandler implements RejectedExecutionHandler { @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 記錄異常 // 報警處理等 System.out.println("error............."); } } // 測試構造的線程池 public static void main(String[] args) { CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor(); // 1.初始化 exec.init(); ExecutorService pool = exec.getCustomThreadPoolExecutor(); for(int i=1; i<100; i++) { System.out.println("提交第" + i + "個任務!"); pool.execute(new Runnable() { @Override public void run() { try { Thread.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("running====="); } }); } // 2.銷毀----此處不能銷毀,因為任務沒有提交執行完,如果銷毀線程池,任務也就無法執行了 // exec.destory(); try { Thread.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } } }復制代碼我們可以看到上述代碼是通過init()方法對ThreadPoolExecutor構造函數進行了一些自定義設置，getCustomThreadPoolExecutor()方法返回init()方法配置的ThreadPoolExecutor對象實例（即線程池引用）
補充：##
ThreadPoolExecutor構造函數有一個參數ThreadFactory threadFactory，前文提到該參數是新建線程的工廠，此處進一步解說該參數。
ThreadFactory是java.util.concurrent包下創建線程工廠的接口，ThreadFactory接口源碼如下：
public interface ThreadFactory { Thread newThread(Runnable r); }復制代碼JDK線程池：Executors.newSingleThreadExecutor、``Executors.newFixedThreadPool等由一個ThreadFactory來創建新的線程，默認情況下為Executors.defaultThreadFactory()，我們可以采用自定義的ThreadFactory工廠，增加對線程創建與銷毀等更多的控制（比如上述代碼中的內部類CustomThreadFactory即為新建線程的模板）
此處簡單提及一下，讀者欲了解更多內容可以參考以下文章
-
（1）guojuanjun.blog.51cto.com/277646/6509…
-
（2）ifeve.com/customizing…
八、擴展ThreadPoolExecutor##
九、源碼視角##
從源碼視角分析Executors、ThreadPoolExecutor、ExecuteService、Executor之間的關系，此處簡單提及，讀者可查看下一節“參考資料”以了解相關內容
- （1）Executors
從Java5開始新增了Executors類，它有幾個靜態工廠方法用來創建線程池，這些靜態工廠方法返回一個ExecutorService類型的值，此值即為線程池的引用。
- （2）Executor
Executor是一個接口，里面只有一個方法
public interface Executor { void execute(Runnable command); }復制代碼- （3）ExecuteService
ExecuteService也是一個接口，其定義如下：
public interface ExecutorService extends Executor {...}
- （4）ThreadPoolExecutor繼承AbstractExecutorService，AbstractExecutorService實現ExecutorService接口
public class ThreadPoolExecutor extends AbstractExecutorService {...}
public abstract class AbstractExecutorService implements ExecutorService {...}
十、ExecutorService的生命周期##
在本文最開始的那個示例中，有一句代碼，如下：
executor.shutdown();
該語句并不是終止線程的運行，而是禁止在這個executor中添加新的任務，下文描述了該語句對于ExecutorService的意義。
十一、參考資料##
本文僅簡單闡述了Java并發中關于Executors及ThreadPoolExecutor的內容，此處貼出一些優質文章以供讀者閱覽
*（1）blog.csdn.net/xiamizy/art…
*（2）www.cnblogs.com/dolphin0520…
*（3）www.cnblogs.com/yezhenhan/a…
*（4）www.cnblogs.com/guguli/p/51…
給大家推薦一個iOS技術微信交流群!群內提供數據結構與算法、底層進階、swift、逆向、底層面試題整合文檔等免費資料！由于群里已到達200人 可加我微信邀請大家進群
轉載于:https://juejin.im/post/5d0b38495188257f9a7678a6
總結
以上是生活随笔為你收集整理的多线程池、饱和策略详解的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： sql 181. 超过经理收入的员工
- 下一篇： SQL表之间复制数据、选出随机几条数据、