一個老生常談的話題：線程的創(chuàng)建及銷毀是非常消耗時間及資源的，所以線程應該交由線程池去執(zhí)行。

ThreadPoolExecutor構造說明及常用方法

ThreadPoolExecutor提供了很多構造方法，這里主要說以下這個，其他構造方法都是基于此方法：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

• 參數int corePoolSize：核心線程池大小，不能小于0；
• 參數int maximumPoolSize：最大線程池大小(核心線程數+臨時線程數)，不能小于等于0且不能小于corePoolSize參數，線程池的大小會在corePoolSize與maximumPoolSize之間動態(tài)變化；
• 參數long keepAliveTime：臨時線程(非核心線程，核心線程空閑之后不會被銷毀)最大空閑時間，不能小于0；
• 參數TimeUnit unit：keepAliveTime參數的時間單位；
• 參數BlockingQueue<Runnable> workQueue：線程等待隊列；
• 參數ThreadFactory threadFactory：線程創(chuàng)建工廠；
• 參數RejectedExecutionHandler handler：拒絕策略；

下面是構造ThreadPoolExecutor的參數校驗：

其他常用方法：

• int prestartAllCoreThreads()：根據線程核心數，啟動所有的核心線程，讓它們全部處于空閑狀態(tài)等待工作
• boolean prestartCoreThread()：隨機啟動一個核心線程，當所有核心線程都已經啟動的時候，返回false(表示未啟動到有效資源)；
• void allowCoreThreadTimeOut(boolean value)：設置是否允許臨時線程永久空閑存活，設置true后，keepAliveTime參數失效；
• boolean allowsCoreThreadTimeOut()：獲取是否允許臨時線程永久空閑存活；
• void shutdown()：銷毀線程池，此前還存在于線程池的任務依然會被執(zhí)行；
• List<Runnable> shutdownNow()：銷毀線程池，與shutdown()的區(qū)別是，未執(zhí)行完成的線程會被標記為中斷；

參數舉例說明：

1、當創(chuàng)建一個任務并交給線程池執(zhí)行執(zhí)行后，會從消耗一個核心線程資源，當該任務執(zhí)行完畢之后，會變成一個空閑線程，另外，如果核心線程數足以支撐任務運行，即使有核心空閑線程，依然會創(chuàng)建一個新的核心線程資源；

2、當加入線程池的任務超過核心線程數，會進入workQueue中等待執(zhí)行；

3、當加入線程池的任務超過核心線程數，溢出的任務數如果超過等待隊列長度，會直接創(chuàng)建大于corePoolSize且小于maximumPoolSize的線程資源用于執(zhí)行任務，超過核心線程數的部分稱為臨時線程；

4、keepAliveTime：當線程池被擴大到corePoolSize與maximumPoolSize之間之后，當執(zhí)行完所有任務后，不再有新的任務進來，超出corePoolSize的部分會作為空閑線程，會在keepAliveTime指定時間后，線程池大小縮小為corePoolSize，如果設置allowCoreThreadTimeOut(true),則空閑線程一直不會銷毀；

5、BlockingQueue<Runnable> workQueue：用于保存等待執(zhí)行的任務的阻塞隊列接口，java提供了以下實現類：

線程池中，BlockingQueue是典型的"生產者消費者"模型：生產者調用插入元素的方法(add/offer/put等方法)，消費者調用移除元素的方法(remove/poll/take等方法)，其中，生產者插入和消費者移除都有阻塞和非阻塞的方法，offer/poll是非阻塞方法，put/take是阻塞方法(隊列后面單獨再寫)；

ThreadPoolExecutor提交任務采用的是不阻塞的offer方法，從隊列中獲取任務采用的是阻塞的take方法，正是因為ThreadPoolExecutor使用了不阻塞的offer方法，所以當隊列容量已滿，線程池會去創(chuàng)建新的臨時線程，去處理隊列中溢出的任務。

• ArrayBlockingQueue：是一個基于數組結構的有界阻塞隊列，此隊列按 FIFO（先進先出）原則對元素進行排序，允許迭代中修改隊列(刪除元素時會更新迭代器)，當然，一般也不會有場景需要迭代阻塞隊列；
• LinkedBlockingQueue：一個基于鏈表結構的默認無界雙端阻塞隊列，雙端意味著可以像普通隊列一樣FIFO(先進先出)，可以以像棧一樣FILO(先進后出)。吞吐量通常要高于ArrayBlockingQueue(如無需在迭代中修改元素，優(yōu)于ArrayBlockingQueue，同時記得指定隊列長度，提供了一個可選有界的構造函數，而在未指明容量時，容量默認為Integer.MAX_VALUE)；
• SynchronousQueue：一個不存儲元素的阻塞隊列，所謂不存儲元素，正如名字描述，同步隊列，每個插入操作必須等到另一個線程調用移除操作，否則插入操作一直處于阻塞狀態(tài)，即生產一個，就消費一個，因此它也被認為是無界的隊列，吞吐量通常要高于LinkedBlockingQueue。同時，SynchronousQueue不能遍歷，因為它沒有元素可以遍歷；
• DelayQueue：使用二叉堆實現的優(yōu)先級阻塞隊列，通過執(zhí)行時延從隊列中提取任務，時間沒到任務取不出來，可用于實現有延時場景的需求；
• PriorityBlockingQueue：使用二叉堆實現的優(yōu)先級阻塞隊列，可以實現Comparable接口也可以提供Comparator來對隊列中的元素進行比較，跟時間沒有任何關系，僅僅是按照優(yōu)先級取任務，如果有根據某個規(guī)則優(yōu)先取出任務的場景，適用；

綜上，需要根據每個隊列的特點選擇適合自己場景的隊列。

6、RejectedExecutionHandler handler：當隊列和線程池都滿了(從上面可以看出，一個線程池可以容納的的最大任務數是maximumPoolSize+隊列長度，因此使用無界隊列會造成隊列永不會滿的情況，一旦控制不好就容易出現OOM)，說明線程池處于飽和狀態(tài)，那么必須采取一種策略處理提交的新任務。

1、AbortPolicy：直接拋出異常；

2、CallerRunsPolicy：只用調用者所在線程來運行任務；

3、DiscardOldestPolicy：丟棄隊列里最近的一個任務，并執(zhí)行當前任務；

4、DiscardPolicy：不處理，丟棄掉；

5、也可以根據應用場景需要來實現RejectedExecutionHandler接口自定義策略。如記錄日志或持久化不能處理的任務；

注：AbortPolicy/CallerRunsPolicy/DiscardOldestPolicy/DiscardPolicy都是ThreadPoolExecutor的內部類，如下：

所以，實例化需要用內部類的方式，如下：

AbortPolicy abortPolicy = new ThreadPoolExecutor.AbortPolicy();

自定義RejectedExecutionHandler拒絕策略示例：

public class RejectedExecutionHandlerTest implements RejectedExecutionHandler {

	@Override
	public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
		// 啥邏輯都沒有,表示忽略,比如DiscardPolicy
		try {
			executor.getQueue().put(r);
		} catch (InterruptedException e) {
			// 	TODO 記錄錯誤日志等
			e.printStackTrace();
		}
		
	}
}

ThreadPoolExecutor的特點總結起來就是以下4點：

1、當有任務提交的時候，會創(chuàng)建核心線程去執(zhí)行任務（即使有核心線程空閑）；

2、當核心線程數達到corePoolSize時，后續(xù)提交的都會進BlockingQueue中排隊；

3、當BlockingQueue滿了(offer失敗)，就會創(chuàng)建臨時線程(臨時線程空閑超過一定時間后，會被銷毀)，其中臨時線程最大數量=maximumPoolSize - corePoolSize，空閑最大時間由keepAliveTime控制，如果設置allowCoreThreadTimeOut(true)，臨時線程永不銷毀；

4、當線程總數達到maximumPoolSize 時，后續(xù)提交的任務都會被RejectedExecutionHandler拒絕。

為什么強制要求使用ThreadPoolExecutor創(chuàng)建線程池

在Executor類中，提供了以下場景的創(chuàng)建線程池的方法：

1、newCachedThreadPool，特點：必要時創(chuàng)建新線程，空閑線程會保留60s；

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>(),
                                      threadFactory);
    }

根據上面的ThreadPoolExecutor，允許的創(chuàng)建線程數量為 Integer.MAX_VALUE，可能會創(chuàng)建大量的線程，從而導致 OOM，還無法指定拒絕策略。

2、newFixedThreadPool，特點：包含固定的線程數，空閑線程會被一直保留；

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }


public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

允許的請求隊列長度為 Integer.MAX_VALUE，可能會堆積大量的請求，從而導致 OOM，還無法指定拒絕策略。

3、newSingleThreadExecutor，特點：只有一個線程的池，順序執(zhí)行每一個提交的任務；

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }



public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
}

在項目中創(chuàng)建線程池

如果需要在項目中創(chuàng)建線程池，那么應該將它設置成單例模式。

示例，用枚舉的方式創(chuàng)建單例：

// 線程工廠
public class UserThreadFactory implements ThreadFactory {

	// 線程組命名標識
	private final String namePrefix;
	// 線程編號
	private final AtomicInteger nextId = new AtomicInteger(1);

	// 定義線程組名稱，在 jstack 問題排查時，非常有幫助
	UserThreadFactory(String whatFeaturOfGroup) {
		namePrefix = "From UserThreadFactory's " + whatFeaturOfGroup + "-Worker-";
	}

	@Override
	public Thread newThread(Runnable task) {
		String name = namePrefix + nextId.getAndIncrement();
		Thread thread = new Thread(task, name);
		return thread;
	}

}


// 單例線程池
public enum ThreadPoolEnum {

	INSTANCE;
	
	private ThreadPoolExecutor threadPoolExecutor;

	// 枚舉的特性,在JVM中只會被實例化一次
	private ThreadPoolEnum() {
		threadPoolExecutor = new ThreadPoolExecutor(2, 
				4, 10, TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>(), 
				new UserThreadFactory("first-threadPool"),
				new ThreadPoolExecutor.DiscardPolicy());
	}
	
	public ThreadPoolExecutor getInstance() {
        return threadPoolExecutor;
    }
	
	public static void main(String[] args) throws InterruptedException {
		ThreadPoolExecutor executor1 = ThreadPoolEnum.INSTANCE.getInstance();
		ThreadPoolExecutor executor2 = ThreadPoolEnum.INSTANCE.getInstance();
		System.out.println(executor1 == executor2);
		Runnable r = () -> {
			System.out.println("新創(chuàng)建的線程任務,交由線程池執(zhí)行");
			System.out.println(Thread.currentThread().getName());
		};
		executor1.execute(r);
		// 銷毀線程池
		executor1.shutdown();
	}
	
}

執(zhí)行結果：

在Spring中創(chuàng)建線程池

@Configuration
public class ThreadPoolConfig {

	// 線程池的參數配置可由外部配置引入
	@Bean(destroyMethod = "shutdown")
	public ThreadPoolExecutor initThreadPoolExecutor() {
		return new ThreadPoolExecutor(2, 
				4, 10, TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>(), 
				new ThreadPoolExecutor.DiscardPolicy());
	}
	
}

總結

以上為個人經驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論