Java并發框架視頻之Executor

更新時間：2020-06-04 16:14:05 來源：動力節點瀏覽2202次

隨著當今處理器中可用的核心數量的增加，隨著對實現更高吞吐量的需求的不斷增長，多線程API變得非常流行。Java提供了自己的多線程框架，稱為Executor框架.

Java并發框架視頻之Executor

1.Executor框架是什么？

Executor框架包含一組用于有效管理工作線程的組件。Executor API通過Executors將任務的執行與要執行的實際任務解耦。這是生產者-消費者模式的一種實現。

java.util.concurrent.Executors提供了用于創建工作線程的線程池的工廠方法。

為了使用Executor框架，我們需要創建一個線程池并提交任務給它以供執行。Executor框架的工作是調度和執行已提交的任務并從線程池中拿到返回的結果。

浮現于腦海中的一個基本的問題是，當我們創建java.lang.Thread的對象或調用實現了Runnable/Callable接口來達到的程序的并行性時，為什么需要線程池？

答案來源于兩個基本面：

為新任務創建新的線程會存在額外的線程創建以及銷毀的開銷。管理這些線程的生命周期會明顯增加CPU的執行時間。

不進行任何限制地為每個進程創建線程會導致創建大量線程。這些線程會占用大量內存并引起資源的浪費。當一個線程利用完CPU的時間片后另一個線程即將利用CPU的時間片時，CPU會花費大量的時間來切換線程的上下文。

所有的這些因素都會導致系統的吞吐量下降。線程池通過保持線程一直存活并重用這些線程來克服這個問題。當提交到線程池中的任務多于正在執行的線程時，那些多余的任務將被放到隊列中。一旦執行任務的線程有空閑的了，它們會從隊列中取下一個任務來執行。對于JDK提供的現成的executors此任務隊列基本是無界的。

2.Executors的類型

現在我們已經了解了executors是什么，讓我們來看看不同類型的executors。

(1)SingleThreadExecutor

此線程池executor只有一個線程。它用于以順序方式的形式執行任務。如果此線程在執行任務時因異常而掛掉，則會創建一個新線程來替換此線程，后續任務將在新線程中執行。

ExecutorService executorService=Executors.newSingleThreadExecutor()

(2)FixedThreadPool(n)

顧名思義，它是一個擁有固定數量線程的線程池。提交給executor的任務由固定的n個線程執行，如果有更多的任務，它們存儲在LinkedBlockingQueue里。這個數字n通常跟底層處理器支持的線程總數有關。

ExecutorService executorService=Executors.newFixedThreadPool(4);

(3)CachedThreadPool

該線程池主要用于執行大量短期并行任務的場景。與固定線程池不同，此線程池的線程數不受限制。如果所有的線程都在忙于執行任務并且又有新的任務到來了，這個線程池將創建一個新的線程并將其提交到executor。只要其中一個線程變為空閑，它就會執行新的任務。如果一個線程有60秒的時間都是空閑的，它們將被結束生命周期并從緩存中刪除。

但是，如果管理得不合理，或者任務不是很短的，則線程池將包含大量的活動線程。這可能導致資源紊亂并因此導致性能下降。

ExecutorService executorService=Executors.newCachedThreadPool();

(4)ScheduledExecutor

當我們有一個需要定期運行的任務或者我們希望延遲某個任務時，就會使用此類型的executor。

ScheduledExecutorService scheduledExecService=Executors.newScheduledThreadPool(1);

可以使用scheduleAtFixedRate或scheduleWithFixedDelay在ScheduledExecutor中定期的執行任務。

scheduledExecService.scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit)
scheduledExecService.scheduleWithFixedDelay(Runnable command,long initialDelay,long period,TimeUnit unit)

這兩種方法的主要區別在于它們對連續執行定期任務之間的延遲的應答。

scheduleAtFixedRate：無論前一個任務何時結束，都以固定間隔執行任務。

scheduleWithFixedDelay：只有在當前任務完成后才會啟動延遲倒計時。

3.對于Future對象的理解

可以使用executor返回的

java.util.concurrent.Future對象訪問提交給executor的任務的結果。Future可以被認為是executor對調用者的響應。

Future&lt;String&gt;result=executorService.submit(callableTask);

如上所述，提交給executor的任務是異步的，即程序不會等待當前任務執行完成，而是直接進入下一步。相反，每當任務執行完成時，executor在此Future對象中設置它。

調用者可以繼續執行主程序，當需要提交任務的結果時，他可以在這個Future對象上調用.get()方法來獲取。如果任務完成，結果將立即返回給調用者，否則調用者將被阻塞，直到executor完成此操作的執行并計算出結果。

如果調用者不能無限期地等待任務執行的結果，那么這個等待時間也可以設置為定時地。可以通過Future.get(long timeout，TimeUnit unit)方法實現，如果在規定的時間范圍內沒有返回結果，則拋出TimeoutException。調用者可以處理此異常并繼續執行該程序。

如果在執行任務時出現異常，則對get方法的調用將拋出一個ExecutionException。

對于Future.get()方法返回的結果，一個重要的事情是，只有提交的任務實現了

java.util.concurrent.Callable接口時才返回Future。如果任務實現了Runnable接口，那么一旦任務完成，對.get()方法的調用將返回null。

另一個關注點是Future.cancel(boolean mayInterruptIfRunning)方法。此方法用于取消已提交任務的執行。如果任務已在執行，則executor將嘗試在mayInterruptIfRunning標志為true時中斷任務執行。

4.Example:創建和執行一個簡單的Executor

我們現在將創建一個任務并嘗試在fixed pool executor中執行它：

public class Task implements Callable&lt;String&gt;{
private String message;
public Task(String message){
this.message=message;
}

?Override
public String call()throws Exception{
return"Hello"+message+"!";
}
}

Task類實現Callable接口并有一個String類型作為返回值的方法。這個方法也可以拋出Exception。這種向executor拋出異常的能力以及executor將此異常返回給調用者的能力非常重要，因為它有助于調用者知道任務執行的狀態。

現在讓我們來執行一下這個任務：

public class ExecutorExample{
public static void main(String[]args){
Task task=new Task("World");
ExecutorService executorService=Executors.newFixedThreadPool(4);
Future&lt;String&gt;result=executorService.submit(task);
try{
System.out.println(result.get());
}catch(InterruptedException|ExecutionException e){
System.out.println("Error occured while executing the submitted task");
e.printStackTrace();
}

executorService.shutdown();
}
}

我們創建了一個具有4個線程數的FixedThreadPool executors，因為這個demo是在四核處理器上開發的。如果正在執行的任務執行大量I/O操作或花費較長時間等待外部資源，則線程數可能超過處理器的核心數。

我們實例化了Task類，并將它提交給executors執行。結果由Future對象返回，然后我們在屏幕上打印。

讓我們運行ExecutorExample并查看其輸出：

Hello World!

正如所料，任務追加了問候語Hello并通過Future object返回結果。

最后，我們調用executorService對象上的shutdown來終止所有線程并將資源返回給OS。

.shutdown()方法等待executor完成當前提交的任務。但是，如果要求是立即關閉executor而不等待，那么我們可以使用.shutdownNow()方法。

任何待執行的任務都將結果返回到java.util.List對象中。

我們也可以通過實現Runnable接口來創建同樣的任務：

public class Task implements Runnable{
private String message;
public Task(String message){
this.message=message;
}

public void run(){
System.out.println("Hello"+message+"!");
}
}

當我們實現Runnable時，這里有一些重要的變化。

無法從run()方法得到任務執行的結果。因此，我們直接在這里打印。

run()方法不可拋出任何已受檢的異常。

Java并發框架視頻之Executor