引言

本篇博客介紹通過“執(zhí)行任務”的機制來設計應用程序時需要掌握的一些知識。所有的內(nèi)容均提煉自《Java并發(fā)編程實戰(zhàn)》中第六章的內(nèi)容。

大多數(shù)并發(fā)應用程序都是圍繞“任務執(zhí)行”來構(gòu)造的：任務通常是一些抽象的且離散的工作單元。

當圍繞“任務執(zhí)行”來設計應用程序結(jié)構(gòu)時，第一步，就是要找出清晰的任務邊界。在理想情況下，各個任務之間是相互獨立的：任務并不依賴于其他任務的狀態(tài)、結(jié)果或邊界效應。

大多數(shù)服務器應用程序提供了一種自然的任務邊界選擇方式：以獨立的客戶請求為邊界。

不好的例子：串行與“為每個任務創(chuàng)建線程”

在書中6.1節(jié)，介紹了由最簡單的串行執(zhí)行任務到為每個任務創(chuàng)建一個線程這兩種執(zhí)行任務的方式。應該說這兩種方式都是不可取的。

這一節(jié)主要是為了引出下一節(jié)介紹的“任務執(zhí)行框架”。

其中“串行執(zhí)行任務”的缺點是在一般的服務器應用程序中，無法提高吞吐率或快速響應性。

而“為每個任務創(chuàng)建線程”的方式的問題在于可能導致：高性能開銷、高資源消耗、影響穩(wěn)定性。

【重點】在工作或面試中也會遇到這個極富針對性的問題，即大量創(chuàng)建線程會存在哪些問題？

1、高性能開銷：創(chuàng)建和銷毀都需要一定的代價，創(chuàng)建過程需要時間，延遲處理請求，也需要jvm和操作系統(tǒng)提供一些輔助操作。

2、高資源消耗：活躍的線程會消耗系統(tǒng)資源，尤其是內(nèi)存。當可運行的線程數(shù)量多余可用處理器的數(shù)量，那么會有大量空閑的線程占用內(nèi)存，不僅給垃圾回收帶來壓力，在競爭CPU的時候還將產(chǎn)生額外的性能開銷。

3、影響穩(wěn)定性：大量線程占用內(nèi)存，內(nèi)存不足，導致可能拋出OutOfMemoryError，系統(tǒng)崩潰。

線程數(shù)量的限制

書中在這里簡單引出一個概念：穩(wěn)定性。

根據(jù)前后文的聯(lián)系，這里具體指的是：應用程序不會因為線程過多而拋出OutOfMemoryError異常。

為了達到這種穩(wěn)定性，在可創(chuàng)建線程數(shù)量上存在一個限制。這個限制受平臺以及多個因素影響，包括JVM啟動參數(shù)、Thread構(gòu)造函數(shù)中請求的棧大小、底層操作系統(tǒng)對線程的限制等。例如，在32位機器上，其中一個主要的限制因素是線程棧的地址空間。每個線程都維護兩個執(zhí)行棧，一個用于Java代碼，另一個用于原生代碼。

通常，JVM在默認情況下會生成一個復合棧，大小約0.5M~1M（這個值可以通過JVM標志 -Xss或通過Thread的構(gòu)造函數(shù)來修改），那么：線程數(shù)量 ≈ 2^32(bit) / 0.5(MB) ≈幾千或幾萬。

因此，在一定范圍內(nèi)，增加線程可以提高系統(tǒng)的吞吐率，但如果超出這個范圍，再創(chuàng)建更多的線程只會降低程序的執(zhí)行速度。

Executor接口

public?interface?Executor?{??void?execute(Runnable?command);?? }??

Executor是一個非常簡單的接口，只有一個execute(Runnable) 方法，它是其他的靈活且強大的異步任務框架的基礎。通過這種方式，用Runnable來表示任務，可以將任務的提交過程與執(zhí)行過程解耦。

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

什么是執(zhí)行策略？

執(zhí)行策略，定義了任務執(zhí)行的“what、where、when、how”等方面，主要是描述根據(jù)不同的資源而選擇不同的執(zhí)行方式，一個最優(yōu)執(zhí)行策略應當是與硬件資源最匹配的。

線程池

先來看一下四種常用線程池的創(chuàng)建：

ExecutorService?newFixedThreadPool?=?Executors.newFixedThreadPool(10);?? ExecutorService?newCachedThreadPool?=?Executors.newCachedThreadPool();?? ScheduledExecutorService?newScheduledThreadPool?=?Executors.newScheduledThreadPool(10);?? ExecutorService?newSingleThreadExecutor?=?Executors.newSingleThreadExecutor();

其中：ExecutorService extends Executor，ScheduledExecutorService extends ExecutorService 。?

1、newFixedThreadPool(int) ：創(chuàng)建一個定額線程池，每提交一個任務創(chuàng)建一個線程，達到數(shù)量限制后不再增加，這時線程池的規(guī)模將不再變化（如果某個線程由于發(fā)生了未預期的異常而結(jié)束，那么線程池會補充一個新的線程）

2、NewCachedThreadPool() : 創(chuàng)建一個可緩存的線程池，線程池的規(guī)模不存在任何限制，當線程多余任務時，回收空閑線程；當任務增加時，創(chuàng)建新線程。

3、NewSingleThreadExecutor：單線程的Executor，如果這個線程異常結(jié)束，會創(chuàng)建另一個線程來替代。NewSingleThreadExecutor能確保依照任務在隊列中的順序串行執(zhí)行（例如FIFO、LIFO、優(yōu)先級）。

4、NewScheduleThreadPool：創(chuàng)建一個固定長度的線程池，而且以延遲或定時的方式來執(zhí)行任務，類似于Timer。

Executor的生命周期

?JVM只有在所有（非守護）線程全部終止后才會退出，無法正確地關閉Executor，JVM將無法結(jié)束。

Executor以異步的方式來執(zhí)行任務，導致了提交任務的狀態(tài)不是立即可見的，即有些任務可能已經(jīng)完成，有些可能正在執(zhí)行，還有些可能正在隊列中等待執(zhí)行。

ExecutorSevice接口就是為了解決執(zhí)行服務的生命周期問題，擴展了Executor接口。它添加了一些用于聲明周期管理的方法（同時還有一些用于任務提交的便利方法）：

public?interface?ExecutorService?extends?Executor?{??void?shutdown();??List<Runnable>?shutdownNow();??boolean?isShutdown();??boolean?isTerminated();??boolean?awaitTermination(long?timeout,?TimeUnit?unit)??throws?InterruptedException;??//?......其他用于任務提交的便利方法?? }??

這五個方法是聲明周期管理的方法，其余的都是與任務提交相關的方法，比如，可以提交比較大的集合Callable對象的方法：

invokeAll(Collection<? extends Callable<T>> tasks)

【重點】ExecutorService的三種狀態(tài)：運行、關閉、已終止 。

ExecutorService在初始創(chuàng)建時處于運行狀態(tài)。shutdown()方法將執(zhí)行平緩的關閉過程：不再接受新的任務，同時等待已經(jīng)提交的任務執(zhí)行完成——包括那些還未開始執(zhí)行的任務。shutdownNow()方法將執(zhí)行粗暴的關閉方式：它將嘗試取消所有運行中的任務，并且不再啟動隊列中尚未開始執(zhí)行的任務。

延遲任務與周期任務

Timer類負責管理延遲任務以及周期任務，但它本身存在缺陷，因此通常要用ScheduleThreadPoolExecutor的構(gòu)造函數(shù)或newScheduleThreadPool工廠方法來創(chuàng)建該類對象。

Timer的缺陷在于，Timer在執(zhí)行所有定時任務時只會創(chuàng)建一個線程。如果某個任務的執(zhí)行時間過長，那么將破壞其他TimerTask的定時精確性。

Timer還有一個問題就是，Timer線程不會捕獲異常，當TimerTask拋出未檢查異常時將終止定時線程。Timer也不會恢復線程的執(zhí)行，而是會錯誤地任務整個Timer都被取消了。這就造成：已經(jīng)被調(diào)度但尚未執(zhí)行的TimerTask將不會再執(zhí)行，新的任務也不會被調(diào)度。稱之為“線程泄漏”。

【重點】生命周期小結(jié)

Runnable和Callable等任務的生命周期：創(chuàng)建、提交、開始、完成、取消。

Future表示的就是一個任務的生命周期。

Thread的生命周期：創(chuàng)建、就緒、運行、阻塞、死亡（或結(jié)束）。

ExecutorService的生命周期（因為它繼承自Executor，因此也是Executor的生命周期）：創(chuàng)建、運行、關閉、已終止。

Callable與Future

callable

Runnable有一個局限性是沒有返回值，也沒辦法拋出受檢異常。對于某些異步獲得結(jié)果的任務無法勝任，Callable應運而生。

它是Runnable的升級版，既可以使用Callable<Void>來達到Runnable一樣的效果，同時也可以使用Callable<T> 來指定返回結(jié)果。

創(chuàng)建Callable的方式有兩種：構(gòu)造函數(shù)、靜態(tài)的封裝方法。

Callable<String>?callableTask?=?new?Callable<String>()?{??@Override??public?String?call()?throws?Exception?{??return?"this?is?a?callable?task....";??}?? };?

Java 8 style：

Callable<String>?callableTask?=?()?->?{??return?"this?is?a?callable?task....";?? };

靜態(tài)方法：Executors.callable(Runnable task, T result)：

Callable<String>?call?=?Executors.callable(()?->?{System.out.println("this?is?a?runnable?task..."); },?"done!");

Future

future表示一個任務的生命周期。主要提供了一些方法用于判斷任務處于哪個階段，還可以獲取任務的結(jié)果甚至是取消任務。它本身還有一層隱含意義是，任務的生命周期只能前進，不能后退，當一個任務處于“完成”狀態(tài)，就永遠停留在“完成”狀態(tài)上。這一點和ExecutorService的生命周期一樣。

Future接口：

public?interface?Future<V>?{??boolean?cancel(boolean?mayInterruptIfRunning);??boolean?isCancelled();??boolean?isDone();??V?get()?throws?InterruptedException,?ExecutionException;??V?get(long?timeout,?TimeUnit?unit)??throws?InterruptedException,?ExecutionException,?TimeoutException;?? }

創(chuàng)建Future的方式通常是使用ExecutorService的submit()方法獲取返回值。如果想通過構(gòu)造器的方式顯式地創(chuàng)建一個任務的生命周期管理對象，可以使用FutureTask。

FutureTask<String>?runnFutureTask?=?new?FutureTask<String>(runnable,?"done!");?? FutureTask<String> callFutureTask = new FutureTask<>(callable);

FutureTask類實現(xiàn)了Runnable和Future兩個接口。

（說明：FutureTask是Java 5加入的類，Java 6又為它補充了一個新的RunnableFuture接口，Runnable接口和Future接口被提升到了RunnableFuture接口上，這更像是一種重構(gòu)手段，我個人認為在實際開發(fā)中用途可能不及直接使用FutureTask）

由于FutureTask實現(xiàn)了Runnable接口，因此可以將它提交給Executor來執(zhí)行，或者直接調(diào)用它的run方法。

是的，FutureTask的run()方法可以直接執(zhí)行任務，而不需要什么start。

Future.get

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

異構(gòu)任務并行化存在的局限

A與B兩個完全不同的任務通過并行方式可以實現(xiàn)小幅度的性能提升，但是如果想大幅度的提升存在一定的困難。因此，得出一個結(jié)論是，只有當大量相互獨立且同構(gòu)的任務可以并發(fā)進行處理時，才能體現(xiàn)出真正的性能提升。

CompletionService與它的子類ExecutorCompletionService

CompletionService是Executor與BlockingQueue的融合。

回顧一下BlockingQueue的一些特性：

BlockingQueue接口是Queue的子接口，有兩個最主要的實現(xiàn)，LinkedBlockingQueue（無界隊列）和ArrayBlockingQueue（有界隊列）。take()或poll()方法都是BlockingQueue的取頭元素的方法，唯一不同的是當沒有可用的頭元素時，take會無限期等待（阻塞），poll可以設置一個超時時間，一旦超時，將返回null。

CompletionService是在任務執(zhí)行的功能上加入了隊列的特性，很明顯是用于處理一批允許有返回值的任務。

用法：創(chuàng)建一個CompletionService（ExecutorCompletionService對象）。【ExecutorCompletionService的構(gòu)造器允許我們傳入一個ExecutorService（用于采取不同的執(zhí)行策略）和一個BlockingQueue（該參數(shù)可選，默認LinkedBlockingQueue）】然后可以將一組Callable任務提交給CompletionService來執(zhí)行，然后使用類似隊列操作的take或poll方法來獲取已完成的結(jié)果，這些結(jié)果會在完成時被封裝為Future。

【擴展】ExecutorCompletionService的實現(xiàn)很簡單。首先通過構(gòu)造函數(shù)創(chuàng)建一個BlockingQueue來保存計算結(jié)果，然后當計算完成時，調(diào)用FutureTask的done方法，放入隊列。展開：當提交某個任務時，該任務將首先包裝為一個QueueingFuture，這是FutureTask【回顧：FutureTask實現(xiàn)了Future、Runnable】的一個子類，QueueingFuture改寫了FutureTask的done方法——將結(jié)果放入BlockingQueue中。take和poll方法委托給BlockingQueue方法，這些方法會在得到結(jié)果之前阻塞。

為任務設置時限

有時候，如果某個任務無法在指定時間內(nèi)完成，那么將不再需要它的結(jié)果，此時可以放棄這個任務。Future.get中支持這種需求：當結(jié)果可用時，它將立即返回，如果在指定時限內(nèi)沒有計算出結(jié)果，那么拋出TimeoutException。

在使用時限任務時需要注意，當這些任務超市后應該立即停止，從而避免為繼續(xù)計算一個不再使用的結(jié)果而浪費計算資源。

【使用Future.get為單個任務設置時限，如果希望對一組任務設置計算時限，比如前面介紹的CompletionService，那么可以使用poll方法來設置執(zhí)行時間】

invokeAll方法

ExecutorServie接口中有兩個重載的invokeAll方法：

<T>?List<Future<T>>?invokeAll(Collection<??extends?Callable<T>>?tasks)??throws?InterruptedException;?? <T>?List<Future<T>>?invokeAll(Collection<??extends?Callable<T>>?tasks,??long?timeout,?TimeUnit?unit)??throws?InterruptedException;

invokeAll方法支持將多個任務提交到一個ExecutorService并獲得結(jié)果。invokeAll方法的參數(shù)為一組任務，并返回一組Future。invokeAll按照任務集合中迭代器的順序?qū)⑺械腇uture添加到返回的集合中，從而使調(diào)用者能夠?qū)⒏鱾€Future與其表示的Callable關聯(lián)起來。

當所有任務都執(zhí)行完畢時，或者調(diào)用線程被中斷時，又或者超過指定時限時，invokeAll都會返回。當超過指定時限，任何還未完成的任務都會取消。當invokeAll返回后，每個任務要么正常完成，要么被取消，而客戶端代碼可以調(diào)用get或isCancelled來判斷究竟是何種情況。

第六章小結(jié)

通過圍繞任務執(zhí)行來設計應用程序，可以簡化開發(fā)過程，并有助于實現(xiàn)并發(fā)。

Executor框架將任務提交與執(zhí)行策略解耦開來，同時還支持多種不同類型的執(zhí)行策略。當需要創(chuàng)建線程來執(zhí)行任務時，可以考慮使用Executor。

要想在將應用程序分解為不同的任務時獲得最大的好處，必須定義清晰的任務邊界。某些應用程序中存在著比較明顯的任務邊界，而在其他一些程序中則需要進一步分析才能揭示出粒度更細的并行性。

總結(jié)

以上是生活随笔為你收集整理的Java并发编程实战————Executor框架与任务执行的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。