使用Java Executor框架實現多線程

本文將涵蓋兩個主題：

• 通過實現Callable接口創建線程

• 在Java中使用Executor框架

實現Callable接口

為了創建一段可以在線程中運行的代碼，我們創建了一個類，然后實現了Callable接口。這段代碼完成的任務需要放在call（）函數中。在下面的代碼中，你可以看到Callable task是一個實現Callable接口的類，在函數中完成了將0到4之間的數字相加的任務。

package com.mzc.common.thread; import java.util.concurrent.Callable; /** * <p class="detail"> * 功能: 實現Callable接口 * </p> * * @author Moore * @ClassName Callable task. * @Version V1.0. * @date 2019.12.23 10:06:23 */ public class CallableTask implements Callable<Integer> { /** * Computes a result, or throws an exception if unable to do so. * * @return computed result * @throws Exception if unable to compute a result */ @Override public Integer call() throws Exception { int sum = 0; for (int i = 0; i < 5; i++) { sum += i; } return sum; } } 

在上面的代碼中，你會注意到Callable的參數為Integer。這表明此Callable的返回類型將為Integer。還可以看到調用函數返回了Integer類型。

1、創建線程並運行

下面的代碼顯示了如何創建線程，然后運行它們。

package com.mzc.common.thread; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.FutureTask; public class CallableInterfaceDemo { public static void main(String[] args) { FutureTask<Integer>[] futureList = new FutureTask[5]; for (int i = 0; i <= 4; i++) { Callable<Integer> callable = new CallableTask(); futureList[i] = new FutureTask<Integer>(callable); Thread t = new Thread(futureList[i]); t.start(); } for (int i = 0; i <= 4; i++) { FutureTask<Integer> result = futureList[i]; try { System.out.println("Future Task" + i + ":" + result.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } } } 

為了創建線程，首先我們需要創建一個CallableTask實例，該實例實現CallableI接口，如圖所示：

Callable<Integer> callable = new CallableTask(); 

然后我們需要創建一個FutureTask類的實例，並將Callable Task的實例作為參數傳遞，如下所示：

futureList[i] = new FutureTask<Integer>(callable); 

然后創建一個線程，我們創建一個Thread類的實例，並將FutureTask類的實例作為參數傳遞，如下所示：

Thread t = new Thread(futureList[i]); 

最后，使用start()方法啟動線程。

2、從線程獲取結果

如果是Callables，線程實際上可以返回一個值。為了獲得該值，我們可以在FutureTask實例上調用get（）函數。在我們的代碼中，線程的返回值是0到4之間的數字之和。如下面的代碼片段所示：

FutureTask<Integer> result = futureList[i]; try { System.out.println("Future Task" + i + ":" + result.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } 

線程可能拋出異常，可以使用try catch塊進行處理。

運行結果：

Future Task0:10 Future Task1:10 Future Task2:10 Future Task3:10 Future Task4:10 

小課堂

上面示例中，我們使用了new Thread()方式創建一個線程，但在實際項目中我們很少這樣用，因為每次new的時候都是新建了一個線程對象，並且不能復用，只能釋放線程資源，這種方式性能較差，而且如果不加限制，創建太多的線程會消耗太多系統資源，同時線程太多，如果要定期執行、關閉線程等，都不方便統一管理。

所以我們就需要線程池來管理線程，使用線程池有以下優點：

• 可以復用線程，減少對象創建，就減少系統資源消耗

• 可以控制最大並發線程數，提高系統資源利用率

• 使用線程池可以避免資源競爭，也就減少了阻塞情況

• 創建線程池簡單方便，同時便於統一管理

Executor框架

每次創建線程都是資源密集型的。一個很好的替代方法是提前設置好一些線程，也就是我上面說的線程池，然后將我們的任務分配給這些線程。這就是Executors類和ExecutorService非常有用的地方。

 

image.png

上圖顯示了具有4個線程的線程池。每當我們希望運行任何任務時，都可以將其分配給這些線程。任務完成后，線程將被釋放以執行其他任務。

1、如何使用Executor框架

我們還是先復用上面實現了Callable接口的CallableTask實例，然后我們改造一下創建線程方式，不再使用new Thread() 創建一個線程，而是先創建一個ExecutorService，Executors類具有ExecutorService的多個實現，這兒我們使用Executors類創建大小為4的固定線程池(newFixedThreadPool)。

ExecutorService executors = Executors.newFixedThreadPool(4);

接下來，我們需要將我們的任務提交給ExecutorService，像這樣：

Future<Integer> future = executors.submit(w); 

提交任務后，我們將獲得一個Future對象的實例。Future對象將存儲Task的結果。完整代碼：

package com.mzc.common.thread; import java.util.concurrent.*; /** * <p class="detail"> * 功能: 使用Executor創建多線程 * </p> * * @author Moore * @ClassName Executor demo. * @Version V1.0. * @date 2019.12.23 10:54:31 */ public class ExecutorDemo { public static void main(String[] args) { /** * 創建線程池的6種方式 */ // 1、創建一個定長線程池，可控制線程最大並發數，超出的線程會在隊列中等待。 ExecutorService executors = Executors.newFixedThreadPool(4); // 2、創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程。 // ExecutorService executors = Executors.newCachedThreadPool(); // 3、創建一個定長線程池，支持定時及周期性任務執行。 // ExecutorService executors = Executors.newScheduledThreadPool(4); // 4、創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。 // ExecutorService executors = Executors.newSingleThreadExecutor(); // 5、newWorkStealingPool工作竊取線程池，它是新的線程池類ForkJoinPool的擴展， // 但是都是在統一的一個Executors類中實現，由於能夠合理的使用CPU進行對任務操作（並行操作）， // 所以適合使用在很耗時的任務中 // ExecutorService executors = Executors.newWorkStealingPool(4); // 6、newSingleThreadScheduledExecutor創建線程池同時放入多個線程時，每個線程都會按照自己的調度來執行， // 但是當其中一個線程被阻塞時，其它的線程都會受到影響被阻塞， // 不過依然都會按照自身調度來執行，但是會存在阻塞延遲。 // ExecutorService executors = Executors.newSingleThreadScheduledExecutor(); Future<Integer>[] futures = new Future[5]; Callable<Integer> w = new CallableTask(); try { for (int i = 0; i < 5; i++) { Future<Integer> future = executors.submit(w); futures[i] = future; } for (int i = 0; i < futures.length; i++) { try { System.out.println("Result from Future " + i + ":" + futures[i].get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } } finally { executors.shutdown(); } } } 

2、獲取結果

為了獲得每個任務的結果，我們可以調用Future實例的get（）方法，方法和上面的一樣：

for (int i = 0; i < futures.length; i++) { try { System.out.println("Result from Future " + i + ":" + futures[i].get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } 

運行結果

Result from Future 0:10 Result from Future 1:10 Result from Future 2:10 Result from Future 3:10 Result from Future 4:10 

固定線程池(newFixedThreadPool)的運行情況說明：

• 在上面的示例中，我們創建了一個大小為4的固定線程池。

• 如果我們總共向ExecutorService提交3個任務，那么所有3個任務都將分配給線程池，並且它們將開始執行。

• 如果我們向ExecutorService提交4個任務，那么所有這4個任務將再次分配給線程池，並且它們將開始執行。

• 如果我們向該線程池提交5個任務，只有4個任務將分配給線程池。這是因為線程池的大小為4。僅當釋放池中的線程之一時，才會分配第五個任務。

關閉ExecutorService

當我們不再需要線程時，需要關閉ExecutorService，這樣做能確保JVM不會消耗其他資源。我們可以使用下面這個命令關閉ExecutorService：

executors.shutdown(); 

ExecutorService的關閉操作通常位於``finally''塊中。這是為了確保即使在發生任何異常的情況下，關閉操作始終在代碼的末尾被執行。如果關閉操作不正確，那么如果發生任何異常，則ExecutorService仍將運行並消耗其他JVM資源。

總結

線程池的一些常用方法

• submit()：提交任務，能夠返回執行結果execute+Future

• shutdown()：關閉線程池，等待任務都執行完

• getPoolSize()：線程池當前線程數量

• getActiveCount()：當前線程池中正在執行任務的線程數量

• shutdownNow()：關閉線程池，不等待任務執行完

• getTaskCount()：線程池已執行和未執行的任務總數

• getCompletedTaskCount()：已完成的任務數量

創建線程池的6種方式

1. newFixedThreadPool()：創建一個定長線程池，可控制線程最大並發數，超出的線程會在隊列中等待。

2. EnewCachedThreadPool()：創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程。

3. newScheduledThreadPool()：創建一個定長線程池，支持定時及周期性任務執行。

4. newSingleThreadExecutor()：創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO,* LIFO*,優先級)執行。

5. newWorkStealingPool()：工作竊取線程池，它是新的線程池類ForkJoinPool的擴展，但是都是在統一的一個Executors類中實現，由於能夠合理的使用CPU進行對任務操作（並行操作），所以適合使用在很耗時的任務中。

6. newSingleThreadScheduledExecutor()：newSingleThreadScheduledExecutor創建線程池同時放入多個線程時，每個線程都會按照自己的調度來執行，但是當其中一個線程被阻塞時，其它的線程都會受到影響被阻塞，不過依然都會按照自身調度來執行，但是會存在阻塞延遲。

代碼示例

// 1、創建一個定長線程池，可控制線程最大並發數，超出的線程會在隊列中等待。 ExecutorService executors = Executors.newFixedThreadPool(4); // 2、創建一個可緩存線程池，如果線程池長度超過處理需要，可靈活回收空閑線程，若無可回收，則新建線程。 ExecutorService executors = Executors.newCachedThreadPool(); // 3、創建一個定長線程池，支持定時及周期性任務執行。 ExecutorService executors = Executors.newScheduledThreadPool(4); // 4、創建一個單線程化的線程池，它只會用唯一的工作線程來執行任務，保證所有任務按照指定順序(FIFO, LIFO, 優先級)執行。 ExecutorService executors = Executors.newSingleThreadExecutor(); // 5、newWorkStealingPool工作竊取線程池，它是新的線程池類ForkJoinPool的擴展， // 但是都是在統一的一個Executors類中實現，由於能夠合理的使用CPU進行對任務操作（並行操作）， // 所以適合使用在很耗時的任務中 ExecutorService executors = Executors.newWorkStealingPool(4); // 6、newSingleThreadScheduledExecutor創建線程池同時放入多個線程時，每個線程都會按照自己的調度來執行， // 但是當其中一個線程被阻塞時，其它的線程都會受到影響被阻塞， // 不過依然都會按照自身調度來執行，但是會存在阻塞延遲。 ExecutorService executors = Executors.newSingleThreadScheduledExecutor(); 

**

如果覺得本文覺得還行，可以關注我的個人公眾號：碼之初。給與我更多寫下去的動力，謝謝！碼之初將為您奉獻：

• Java基礎到架構知識全方位涵蓋

• 海量面試資料，從文檔到視頻，從基礎到底層，面試無憂

• 程序猿生活記錄，程序猿的生活也值得關注

• 不定期的福利發放，驚喜就是在不經意間來臨。