Java併發實戰（六） 任務執行

1.任務

把程序抽象成多個任務。

2.現代web程序劃分任務邊界

以獨立的客戶請求爲邊界。就是一個請求一個任務。

3.任務調度策略

3.1串行的執行

糟糕的響應性和吞吐量。

3.2爲每個任務建立一個線程

結論:

• 任務交由子線程處理，提升了響應性和吞吐量。
• 任務處理的代碼必須是線程安全的。

不足:

• 線程生命週期的開銷很是高。
• 資源消耗。可運行線程多於可用處理器的數量，會有線程閒置佔用內存，且大量線程競爭CPU時將產生其餘性能開銷。
• 穩定性。不一樣平臺可建立線程的數量有限制。

4.Java中Executor框架的設計

4.1設計理念

Java提供了Executor框架來執行任務。基於生產者-消費者模式。提交任務就是操做至關於生產者，執行任務的線程至關於消費者。（解耦，削峯）

public interface Executor {
        void execute(Runnable command);
    }
複製代碼

4.2執行策略

任務的提交代碼散佈在整個程序的業務代碼中。
執行策略則統一交由框架處理。

執行策略中定義了任務執行的"What,Where,When,How"等方面，包括：

• 在什麼(What)線程中執行任務?
• 任務按照什麼(What)順序執行(優先級)?
• 有多少個(How Many)任務能併發執行?
• 在隊列中有多少個(How Many)任務在等待執行?
• 系統該怎麼(How)拒絕任務?
• 在任務執行先後,應該進行哪些(What)動做?

經過將任務提交與任務的執行策略分離，有助於在部署階段選擇與可用硬件資源最匹配的執行策略。

4.3線程池

Executor任務執行框架將"爲每個任務分配一個線程"策略編程基於線程池的策略。
類庫提供了一個靈活的線程池及一些有用的默認配置。如newFixedThreadpool。

• Web服務器不會再高負載狀況下失敗。
• 可是任務到達的速度老是超過任務執行的速度,服務器仍有可能耗盡內存。

4.4Executor的生命週期

Executor擴展了ExecutorService接口,添加了一些用於生命週期管理的方法。

public interface ExecutorService extends Executor {
        /**
         * 平緩的關閉過程:再也不接受新任務，等待已經提交的任務執行完成。
         */
        void shutdown();
        
        /**
         * 粗暴的關閉過程:它將嘗試取消全部運行中的任務,不在啓動隊列中還沒有開始執行的任務。
         */
        list<Runnable> shutdownNow();
        
        boolean isShutdown();
        boolean isTerminated();
        boolean awaitTermination(long timeout, TimeUnit unit);
    }
複製代碼

4.5延遲任務和週期任務

JAVA中提供Timer來管理定時任務。

• Timer執行定時任務只會建立一個線程。
• Timer是基於絕對時間的調度機制,對系統時間敏感。
• Timer存在線程泄露問題(Timer不捕獲異常，當拋出一個未檢查異常時線程將終止)。

ScheduledThreadPoolExecutor更優質的管理定時任務。

• 其內部是一個線程池。
• 其很好的解決了Timer的線程泄露問題。

不適用於分佈式環境。

5.找出可利用的並行性

本章提供一些示例來發掘在一個請求中的並行性。

5.1 示例:串行的頁面渲染器

假設頁面 = 文本標籤 + 圖片
以下代碼串行的執行渲染。

public class SingleThreadRenderer {
        void renderPage(CharSequence source) {
            renderText(source);
            List<ImageData> imageData = new ArrayList<ImageData>();
            for (ImageInfo imageInfo : scanForImageInfo(source))
                imageData.add(imageInfo.downloadImage());
            for (ImageData data : imageData)
                renderImage(data);
        }
    }
複製代碼

5.2攜帶結果的任務Callable與Future

Runnable做爲基本的任務表現形式。缺陷:1.無返回值。2.不能拋出一個受檢查異常。

• Callable接口

它是任務更好的抽象,描述了一個任務的返回值和異常。

public interface Callable<V> {
        V call() throws Exception;
    }
複製代碼

• Future接口

它表示一個任務的生命週期,並提供了相應的方法來判斷任務是否已經完成或取消。

public interface Future<V> {
        boolean cancel(boolean mayInterruptIfRunning);
        boolean isCancelled();
        boolean isDone();
        V get() throws Exception;
        V get(long timeout, TimeUnit unit);
    }
複製代碼

5.3示例:使用Future實現頁面渲染器

將渲染過程分解成兩個任務,一個是渲染全部的文本,一個是下載全部圖像。

代碼略。
複製代碼

渲染文本和渲染圖片併發執行。

5.4在異構任務並行化中存在的侷限

上例中通常渲染文本的速度遠遠高於渲染圖片的速度,程序最終和串行執行效率差異不大,代碼確變得更復雜了。

只有大量相互獨立且同構的任務能夠併發進行處理時,才能體現出性能的提高。

5.5CompletionService:Executor與BlockingQueue

提交一組任務,簡單的寫法。

@Test
    public void test() throws Exception{
        ExecutorService executor = Executors.newFixedThreadPool(5);
        List<Future<String>> futures = new ArrayList();
        for (int i=0; i<5; i++){
            final int param = i;
            Future<String> future = executor.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    Thread.sleep(param * 1000);
                    return "result" + param;
                }
            });
            futures.add(future);
        }

        for (int i=4; i>0; i--) {
            System.out.println(futures.get(i).get());
        }
    }
複製代碼

CompletionService將Executor和BlockingQueue的功能融合。你能夠將Callable任務提交給它執行，而後使用相似隊列操做的take和poll方法來得到已完成的結果。

5.6示例:使用CompletionService實現頁面渲染器

書上的示例:略。

@Test
    public void test() throws Exception{
        ExecutorService executor = Executors.newFixedThreadPool(5);
        CompletionService<String> completionService = new ExecutorCompletionService<>(executor);
        for (int i=4; i>0; i--){
            final int param = i;
            completionService.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    Thread.sleep(param * 1000);
                    return "result" + param;
                }
            });
        }
        for (int i=0; i<4; i++) {
            System.out.println(completionService.take().get());
        }
    }
    輸出:
        result1
        result2
        result3
        result4
複製代碼

5.7爲任務設置時限

爲單個任務設置時間。

@Test
    public void singleTaskTest(){
        ExecutorService executor = Executors.newFixedThreadPool(5);
        Future<String> future = executor.submit(new Callable<String>() {
            @Override
            public String call() {
                try {
                    Thread.sleep(2000L);
                }catch (Exception e){
                    e.printStackTrace();
                }
                System.out.println("任務執行完畢...");
                return "singleTask.";
            }
        });
        try {
            System.out.println(future.get(1, TimeUnit.SECONDS));
        }catch (TimeoutException e){
            System.out.println("任務超時...");
            future.cancel(true); // 這句話的是否註銷影響運行狀況,原理未知？
        }catch (InterruptedException e){
            e.printStackTrace();
        }catch (ExecutionException e){
            e.printStackTrace();
        }
    }
複製代碼

5.8示例:陸行預約門戶網站

未多個任務設置超時時間。

6.總結

本章主要是介紹了Java的Executor框架的優勢和一些常見需求。 還有對任務的劃分粒度,要根據業務場景分析任務邊界。