前言

Executor 框架是 Java5 之后引進的，在 Java 5 之后，通過 Executor 來啟動線程比使用 Thread 的 start 方法更好，除了更易管理，效率更好（用線程池實現(xiàn)，節(jié)約開銷）。

Executor 框架不僅包括了線程池的管理，還提供了線程工廠、隊列以及拒絕策略等，Executor 框架讓并發(fā)編程變得更加簡單。

Executor框架的組成：

• 任務（Runnable/Callable）：任務通過Runnable接口或Callable接口進行定義。Runnable接口或Callable接口實現(xiàn)類都可以被 ThreadPoolExecutor執(zhí)行
• 任務的執(zhí)行（Executor）：任務執(zhí)行機制的核心接口 Executor，以及繼承自Executor接口的ExecutorService接口。ThreadPoolExecutor實現(xiàn)了ExecutorService接口
• 異步的計算結果（Future）：Future接口以及Future接口的實現(xiàn)類FutureTask類都可以代表異步計算的結果。當我們把Runnable接口或Callable接口的實現(xiàn)類提交給ThreadPoolExecutor執(zhí)行后就會返回一個Future對象

一、Executor接口

線程池簡化了線程的管理工作， 并且JUC提供了一種靈活的線程池實現(xiàn)來作為Executor框架的一部分。

在Java類庫中，任務執(zhí)行的主要抽象不是Thread而是Executor，Executor只定義了execute一個方法，是最頂層的接口。

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the {@code Executor} implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);

execute方法接受一個Runnable參數(shù)，這個方法定義為在未來的某個時間執(zhí)行傳入的方法，方法的運行可以在一個新的線程，在線程池或者在調用的線程中，該方法無法接收到線程的執(zhí)行結果（當然Runnable接口本身也沒有返回值）。

雖然Executor是個簡單的接口，但它卻為靈活且強大的異步任務執(zhí)行框架提供了基礎，該框架能支持多種不同類型的任務執(zhí)行策略。它提供了一種標準的方法將任務的提交過程與執(zhí)行過程解耦開來，并用Runnable來表示任務。

Executor的實現(xiàn)還提供了對生命周期的支持，以及統(tǒng)計信息收集、應用程序管理機制和性能監(jiān)視等機制。

**Executor基于生產(chǎn)者-消費者模式，提交任務的操作相當于生產(chǎn)者，執(zhí)行任務的線程相當于消費者。**如果要在程序中實現(xiàn)一個生產(chǎn)者-消費者的設計，那么最簡單的方式就是使用Executor。

二、ExecutorService接口

Executor框架使用Runnable作為其基本的任務表示形式。Runnable是一種有很大局限的抽象，它的run方法執(zhí)行任務后不能返回一個值或者拋出一個受檢查的異常。許多任務實際上都是存在延遲的計算——執(zhí)行數(shù)據(jù)庫查詢，從網(wǎng)絡上獲取資源，或者計算某個復雜的功能。對于這些任務，Callable是一種更好的抽象，它認為主入口點(call)將返回一個值，并可能拋出一個異常。

因此引入了ExecutorService，它繼承自Executor，并且在其基礎上增加了很多功能，可以生成用于跟蹤一個或多個異步任務進度的Future的方法，以解決Executor的局限性。

Future表示一個任務的生命周期，并且提供了響應的方法來判斷是否已經(jīng)完成或取消，以及獲取任務的結果和取消任務等。

在Future規(guī)范中包含的隱含意義是任務的生命周期只能前進，不能后退，就像ExecutorService的生命周期一樣。當某個任務完成后，它就永遠停留在完成狀態(tài)。

get方法的行為取決于任務的狀態(tài)（尚未開始、正在運行、已完成）。如果任務已完成那么get會立即返回或者拋出一個Exception，如果任務沒有完成，那么get將阻塞并直到任務完成。如果任務拋出了一場，那么get將該異常封裝為ExecutionException并重新拋出。如果任務被取消，那么get將拋出CancellationException。如果get拋出了ExecutionException，那么可以通過getCause來獲得被封裝的初始異常。

定義了以下方法：

• void shutdown()：啟動有序關機，其中執(zhí)行先前提交的任務，但不接受新任務。如果調用已經(jīng)關閉，則沒有額外的效果。不會阻塞等待先前提交的任務執(zhí)行完成
• List<Runnable> shutdownNow()：嘗試停止所有正在執(zhí)行的任務，停止對等待任務的處理，并返回等待執(zhí)行的任務列表。不會阻塞等待正在執(zhí)行的任務終止只是盡最大努力停止處理正在執(zhí)行的任務之外，沒有任何保證。例如，典型的實現(xiàn)將通過Thread.interrupt取消，因此任何未能響應中斷的任務可能永遠不會終止
• boolean isShutdown()：如果此執(zhí)行器已關閉（調用了關閉方法），則返回true
• boolean isTerminated()：如果關閉后所有任務都已完成，則返回true。注意，除非先調用shutdown或shutdownNow，否則isTerminated永遠不會為真
• boolean awaitTermination(long timeout, TimeUnit unit)：阻塞直到所有任務在關機請求后完成執(zhí)行，或者超時發(fā)生，或者當前線程被中斷，以先發(fā)生的為準
• <T> Future<T> submit(Callablet task)：提交一個帶返回值的任務以供執(zhí)行，并返回表示該任務的掛起結果的Future。Future的get方法將在成功完成任務時返回任務的結果
• <T> Future<T> submit(Runnable task, T result)：提交可運行任務以供執(zhí)行，并返回表示該任務的Future。Future的get方法將在成功完成時返回給定的結果
• Future<?> submit(Runnable task)：提交可運行任務以供執(zhí)行，并返回表示該任務的Future。Future的get方法將在成功完成時返回null
• <T> Listfuture<t> invokeAll(Collection? extends Callable<T> tasks)：執(zhí)行給定的任務，并在所有任務完成時返回保存其狀態(tài)和結果的future列表。每個Future對象的isDone方法都會返回true。請注意，已完成的任務可以正常終止，也可以拋出異常終止。如果在執(zhí)行此操作時修改了給定的集合，則此方法的結果是未定義的
• <T> T invokeAny(Collection? extends Callable<T> tasks)：執(zhí)行給定的任務，如果有成功完成的任務，則返回成功完成的任務的結果(即不拋出異常)。在正常或異常返回時，未完成的任務將被取消。如果在執(zhí)行此操作時修改了給定的集合，則此方法的結果是未定義的。

三、ThreadPoolExecutor類

ThreadPoolExecutor實現(xiàn)了ExecutorService（實際上是繼承了AbstractExecutorService），為了在廣泛的上下文中發(fā)揮作用，該類提供了許多可調參數(shù)和可擴展性掛鉤：

• corePoolSize：指定了線程池中的線程數(shù)量，它的數(shù)量決定了添加的任務是開辟新的線程去執(zhí)行，還是放到workQueue任務隊列中去
• maximumPoolSize：指定了線程池中的最大線程數(shù)量，這個參數(shù)會根據(jù)你使用的workQueue任務隊列的類型，決定線程池會開辟的最大線程數(shù)量
• keepAliveTime：當線程池中空閑線程數(shù)量超過corePoolSize時，多余的線程（救急線程）會在多長時間內被銷毀
• unit：keepAliveTime的單位
• workQueue：任務隊列，被添加到線程池中，但尚未被執(zhí)行的任務；它一般分為直接提交隊列、有界任務隊列、無界任務隊列、優(yōu)先任務隊列幾種
• threadFactory：線程工廠，用于創(chuàng)建線程，一般用默認即可
• handler：拒絕策略；當任務太多來不及處理時，如何拒絕任務

1、狀態(tài)

1. RUNNING(-1<<29)：接受新任務并且處理排隊任務
2. SHUTDOWN(0<<29)：不接受新任務但是處理排隊任務
3. STOP(1<<29)：不接受新任務也不處理排隊任務，同時中斷處理中的任務
4. TYDING(2<<29)：所有任務被終止，工作線程數(shù)為0之后進入該狀態(tài)，并且會執(zhí)行terminated()鉤子函數(shù)
5. TERMINATED(3<<29)：terminated()鉤子函數(shù)執(zhí)行完畢后

狀態(tài)單調地隨時間增加，但不需要達到每個狀態(tài)。

• RUNNING->SHUTDOWN：調用shutdown()
• RUNNING/SHUTDOWN->STOP：調用shutdownNow()
• STOP->TYDING：當隊列和池都為空時
• TIDYING -> TERMINATED：terminated()鉤子函數(shù)執(zhí)行完畢后

ThreadPoolExecutor中對于狀態(tài)的記錄保存在一個AtomicInteger類型的變量中，其中高三位就是用于記錄線程池的狀態(tài)，而低的29位用于記錄線程數(shù)量。

2、Worker

ThreadPoolExecutor中定義了一個私有靜態(tài)類Worker，其繼承自AbstractQueuedSynchronizer類，并實現(xiàn)了Runnable接口。其中維護了線程實例（Thread）、任務實例（Runnable）、線程任務計數(shù)器（long）變量。

這個類適當?shù)財U展了AbstractQueuedSynchronizer，以簡化獲取和釋放圍繞每個任務執(zhí)行的鎖。這可以防止中斷，這些中斷旨在喚醒等待任務的工作線程，而不是中斷正在運行的任務。我們實現(xiàn)了一個簡單的不可重入互斥鎖，而不是使用ReentrantLock，因為我們不希望工作任務在調用setCorePoolSize等池控制方法時能夠重新獲得鎖。此外，為了在線程實際開始運行任務之前抑制中斷，我們將鎖狀態(tài)初始化為負值，并在啟動時(在runWorker中)清除它。

3、擴展

該類還定義了三個protected類型的鉤子函數(shù)：

• beforeExecute：線程池任務運行前執(zhí)行
• afterExecute：線程池任務運行后執(zhí)行
• terminated：線程池退出后執(zhí)行

這幾個方法在ThreadPoolExecutor中為空實現(xiàn)。

四、ForkJoinPool類

Fork/Join框架是Java7提供的一個用于并行執(zhí)行任務的框架，是一個把大任務分割成若干個小任務，最終匯總每個小任務結果后得到大任務結果的框架。

Fork就是把一個大任務切分為若干個子任務并行的執(zhí)行，Join就是合并這些子任務的執(zhí)行結果，最后得到這個大任務的結果。比如計算1+2+…+10000，可以分割成10個子任務，每個子任務分別對1000個數(shù)進行求和，最終匯總這10個子任務的結果。

1、工作竊取算法

ForkJoinPool是運行ForkJoinTasks的ExecutorService。**ForkJoinPool與其他類型的ExecutorService的區(qū)別主要在于使用了工作竊取。工作竊取算法是指某個線程從其他隊列里竊取任務來執(zhí)行。**那么為什么要使用工作竊取算法呢？**假如我們需要做一個比較大的任務，可以把這個任務分割為若干個互不干擾的子任務，為了減少線程間的競爭，把這些子任務分別放到不同的隊列里，并為每個隊列創(chuàng)建一個單獨的線程來執(zhí)行隊列里的任務，線程和隊列一一對應。**比如A線程負責處理A隊列里的任務。但是有的線程會先把自己隊列里的任務干完，而其他線程對應的隊列里還有任務等待處理。干完活的線程與其等著，不如去幫其他線程干活，于是它就去其他線程的隊列里竊取一個任務來執(zhí)行。而這時它們會訪問同一個隊列，所以為了減少竊取任務線程之間的競爭，通常會使用雙端隊列，被竊取任務線程永遠從雙端隊列的頭部拿任務，而竊取任務的線程永遠從雙端隊列的尾部拿任務執(zhí)行。

工作竊取算法的優(yōu)點：充分利用線程進行并行計算，減少了線程間的競爭

工作竊取算法的缺點：在某些情況下還是存在競爭，比如雙端隊列里只有一個任務時。并且該算法會消耗了更多的系統(tǒng)資源，比如創(chuàng)建多個線程和多個雙端隊列。

2、Fork/Join的設計

想要設計一個Fork/Join框架，需要完成兩個步驟：

• 分割任務：需要有一個fork類來把大任務分割成子任務，有可能子任務還是很大，所以還需要不停地分割，直到分割出的子任務足夠小
• 執(zhí)行任務并合并結果：分割的子任務分別放在雙端隊列里，然后幾個啟動線程分別從雙端隊列里獲取任務執(zhí)行。子任務執(zhí)行完的結果都統(tǒng)一放在一個隊列里，啟動一個線程從隊列里拿數(shù)據(jù)，然后合并這些數(shù)據(jù)。

Fork/Join使用兩個類來完成以上兩件事情：

• ForkJoinTask（抽象類）：我們要使用ForkJoin框架，必須首先創(chuàng)建一個ForkJoin任務。它提供在任務中執(zhí)行fork()和join()操作的機制。通常情況下我們不需要直接繼承ForkJoinTask類，只需要繼承它的子類，F(xiàn)ork/Join框架提供了以下兩個子類：
  • RecursiveAction（抽象類）：用于沒有返回結果的任務
  • RecursiveTask（抽象類）：用于有返回結果的任務
• ForkJoinPool：ForkJoinTask需要通過ForkJoinPool來執(zhí)行

任務分割出的子任務會添加到當前工作線程所維護的雙端隊列中，進入隊列的頭部。當一個工作線程的隊列里暫時沒有任務時，它會隨機從其他工作線程的隊列的尾部獲取一個線程。

public class CountTask extends RecursiveTask<Integer> {
	private int start;
	private int end;
	public CountTask(int start, int end) {
		this.start = start;
		this.end = end;
	}
	@Override
	protected Integer compute() {
		int sum = 0;
		// 如果任務足夠小就計算任務
		boolean canCompute = (end - start) <= THRESHOLD;
		if (canCompute) {
			for (int i = start; i <= end; i++) {
				sum += i;
			}
		} else {
			// 如果任務大于閾值，就分裂成兩個子任務計算
			int middle = (start + end) / 2;
			CountTask leftTask = new CountTask(start, middle);
			CountTask rightTask = new CountTask(middle + 1, end);
			// 執(zhí)行子任務
			leftTask.fork();
			rightTask.fork();
			// 等待子任務執(zhí)行完，并得到其結果
			int leftResult = leftTask.join();
			int rightResult = rightTask.join();
			// 合并子任務
			sum = leftResult + rightResult;
		}
		return sum;
	}
	public static void main(String[] args) {
		ForkJoinPool forkJoinPool = new ForkJoinPool();
		CountTask task = new CountTask(1, 4);
		// 執(zhí)行一個任務
		Future<Integer> result = forkJoinPool.submit(task);
		try {
			System.out.println(result.get());
		} catch (InterruptedException e) {
		} catch (ExecutionException e) {
		}
	}
}

RecursiveTask需要實現(xiàn)compute方法，在這個方法里，首先需要判斷任務是否足夠小，如果足夠小就直接執(zhí)行任務。如果不足夠小就必須分割成兩個子任務，每個子任務在調用fork方法時又會進入compute方法。最后使用join方法等待子任務執(zhí)行完成并得到其結果。

ForkJoinTask在執(zhí)行的時候可能會拋出異常，但是我們沒辦法在主線程里直接捕獲異常，所以ForkJoinTask提供了isCompletedAbnormally()方法來檢查任務是否已經(jīng)拋出異常或已經(jīng)被取消了，并且可以通過ForkJoinTask的getException方法來獲取異常。

getException方法返回Throwable對象，如果任務被取消了則返回CancellationException，如果任務沒有完成或者沒有拋出異常則返回null。

3、執(zhí)行原理

ForkJoinPool由ForkJoinTask數(shù)組和ForkJoinWorkerThread數(shù)組組成，F(xiàn)orkJoinTask數(shù)組負責存放程序提交給ForkJoinPool的任務，而ForkJoinWorkerThread數(shù)組負責執(zhí)行這些任務。

當我們調用ForkJoinTask的fork方法時，程序會調用ForkJoinWorkerThread的pushTask方法異步地執(zhí)行這個任務，然后立即返回結果。

pushTask方法把當前任務存放在ForkJoinTask數(shù)組隊列里，然后再調用ForkJoinPool的signalWork反復噶喚醒或創(chuàng)建一個工作線程來執(zhí)行任務。

五、ScheduledThreadPool類

ScheduledThreadPoolExecutor主要用來在給定的延遲后運行任務，或者定期執(zhí)行任務。ScheduledThreadPoolExecutor使用任務隊列DelayQueue封裝了一個PriorityQueue，PriorityQueue會對隊列中的任務進行排序，執(zhí)行所需時間短的放在前面先被執(zhí)行(ScheduledFutureTask的time變量小的先執(zhí)行)，如果執(zhí)行所需時間相同則先提交的任務將被先執(zhí)行(ScheduledFutureTask的squenceNumber變量小的先執(zhí)行)。

1、ScheduledExecutorService

ScheduledThreadPool類繼承自ThreadPoolExecutor，并且實現(xiàn)了ScheduledExecutorService接口。

ScheduledExecutorService接口定義了幾個方法：

1. ScheduleFuture<?> schedule(Runnable command, long delay, TimeUnit unit)：提交在給定延遲后啟用的一次性任務
2. ScheduleFuture schedule(Callable command, long delay, TimeUnit unit)：提交在給定延遲之后啟用的帶返回值的一次性任務
3. ScheduleFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)：提交一個周期性任務，任務開始時進行計時（如果任務執(zhí)行時間過長甚至超過period時間，會導致任務連續(xù)執(zhí)行）
4. ScheduleFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)：提交一個周期性任務，任務執(zhí)行完成之后才進行計時

ScheduledFuture繼承自Delayed接口和Future接口，自己本身沒有定義新的方法。

Delayed接口是一個混合風格的接口，用于標記應該在給定延遲后執(zhí)行的對象。這個接口定義了一個getDelay方法，用于返回剩余的延遲時間。此外這個接口繼承了Comparable接口，意味著此接口的實現(xiàn)必須定義一個compareTo方法，該方法提供與其getDelay方法一致的排序

2、比較Timer

Timer對系統(tǒng)時鐘的變化敏感，ScheduledThreadPoolExecutor不是

Timer只有一個執(zhí)行線程，因此長時間運行的任務可以延遲其他任務。 ScheduledThreadPoolExecutor可以配置任意數(shù)量的線程。 此外，如果你想（通過提供 ThreadFactory），你可以完全控制創(chuàng)建的線程

在TimerTask中拋出的運行時異常會殺死一個線程，從而導致 Timer 死機，即計劃任務將不再運行。ScheduledThreadExecutor不僅捕獲運行時異常，還允許您在需要時處理它們（通過重寫afterExecute方法ThreadPoolExecutor）。拋出異常的任務將被取消，但其他任務將繼續(xù)運行

六、Executors類

Executors是Java中用于創(chuàng)建線程池的工廠類，它提供了一系列的靜態(tài)工廠方法，用于創(chuàng)建不同類型的線程池。這些工廠方法隱藏了線程池的復雜性，使得線程池的創(chuàng)建變得非常簡單。Executors工廠類提供的線程池有以下幾種類型：

• newCachedThreadPool()：CachedThreadPool的corePoolSize 被設置為0，maximumPoolSize被設置為Integer.MAX.VALUE，即它是無界的，這也就意味著如果主線程提交任務的速度高于maximumPool中線程處理任務的速度時，CachedThreadPool會不斷創(chuàng)建新的線程。極端情況下，這樣會導致耗盡 cpu和內存資源
• newFixedThreadPool(int nThreads)：創(chuàng)建一個固定大小的線程池，其中包含指定數(shù)量的線程。線程數(shù)量是固定的，不會自動擴展，即沒有救急線程
• newSingleThreadExecutor()：創(chuàng)建一個單線程的線程池。這個線程池中只包含一個線程，用于串行執(zhí)行任務。適用于需要按順序執(zhí)行任務的場景
• newScheduledThreadPool(int corePoolSize)：創(chuàng)建一個固定大小的線程池，用于定時執(zhí)行任務。線程數(shù)量固定，不會自動擴展。適用于定時執(zhí)行任務的場景
• newSingleThreadScheduledExecutor()：創(chuàng)建一個單線程的定時執(zhí)行線程池。只包含一個線程，用于串行定時執(zhí)行任務
• newWorkStealingPool(int parallelism)：該線程池維護足夠的線程以支持給定的并行級別，并且可以使用多個隊列來減少爭用。并行性級別對應于積極參與或可用參與任務處理的最大線程數(shù)。實際的線程數(shù)可以動態(tài)地增加和減少。工作竊取池不能保證所提交任務的執(zhí)行順序

除此之外還提供了創(chuàng)建ThreadFactory實例，將Runnable實例轉換為Callable實例等方法。

到此這篇關于Java多線程中的Executor框架解析的文章就介紹到這了,更多相關Java的Executor框架內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論