前言

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

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

Executor框架的組成：

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

一、Executor接口

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

在Java類庫中，任務(wù)執(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í)行傳入的方法，方法的運行可以在一個新的線程，在線程池或者在調(diào)用的線程中，該方法無法接收到線程的執(zhí)行結(jié)果（當(dāng)然Runnable接口本身也沒有返回值）。

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

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

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

二、ExecutorService接口

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

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

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

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

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

定義了以下方法：

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

三、ThreadPoolExecutor類

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

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

1、狀態(tài)

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

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

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

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

2、Worker

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

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

3、擴展

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

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

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

四、ForkJoinPool類

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

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

1、工作竊取算法

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

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

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

2、Fork/Join的設(shè)計

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

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

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

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

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

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

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

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

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

3、執(zhí)行原理

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

當(dāng)我們調(diào)用ForkJoinTask的fork方法時，程序會調(diào)用ForkJoinWorkerThread的pushTask方法異步地執(zhí)行這個任務(wù)，然后立即返回結(jié)果。

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

五、ScheduledThreadPool類

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

1、ScheduledExecutorService

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

ScheduledExecutorService接口定義了幾個方法：

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

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

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

2、比較Timer

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

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

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

六、Executors類

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

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

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

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

最新評論