2. 程式人生 > >java併發程式設計--Executor框架以及Executors類的建立執行緒池方法

java併發程式設計--Executor框架以及Executors類的建立執行緒池方法

阿新 • • 發佈：2018-11-11

       Eexecutor作為靈活且強大的非同步執行框架，其支援多種不同型別的任務執行策略，提供了一種標準的方法將任務的提交過程和執行過程解耦開發，基於生產者-消費者模式，其提交任務的執行緒相當於生產者，執行任務的執行緒相當於消費者，並用Runnable來表示任務，Executor的實現還提供了對生命週期的支援，以及統計資訊收集，應用程式管理機制和效能監視等機制。

1.Exexctor簡介

Executor的UML圖：（常用的幾個介面和子類）

Executor：一個介面，其定義了一個接收Runnable物件的方法executor，其方法簽名為executor(Runnable command),

ExecutorService：是一個比Executor使用更廣泛的子類介面，其提供了生命週期管理的方法，以及可跟蹤一個或多個非同步任務執行狀況返回Future的方法

AbstractExecutorService：ExecutorService執行方法的預設實現

ScheduledExecutorService：一個可定時排程任務的介面

ScheduledThreadPoolExecutor：ScheduledExecutorService的實現，一個可定時排程任務的執行緒池

ThreadPoolExecutor：執行緒池，可以通過呼叫Executors以下靜態工廠方法來建立執行緒池並返回一個ExecutorService物件：

2.ThreadPoolExecutor建構函式的各個引數說明

ThreadPoolExecutor方法簽名：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) //後兩個引數為可選引數

引數說明：

corePoolSize：核心執行緒數，如果執行的執行緒少於corePoolSize，則建立新執行緒來執行新任務，即使執行緒池中的其他執行緒是空閒的

maximumPoolSize:最大執行緒數，可允許建立的執行緒數，corePoolSize和maximumPoolSize設定的邊界自動調整池大小：

corePoolSize <執行的執行緒數< maximumPoolSize:僅當佇列滿時才建立新執行緒

corePoolSize=執行的執行緒數= maximumPoolSize：建立固定大小的執行緒池

keepAliveTime:如果執行緒數多於corePoolSize,則這些多餘的執行緒的空閒時間超過keepAliveTime時將被終止

unit:keepAliveTime引數的時間單位

workQueue:儲存任務的阻塞佇列，與執行緒池的大小有關：

  當執行的執行緒數少於corePoolSize時，在有新任務時直接建立新執行緒來執行任務而無需再進佇列

  當執行的執行緒數等於或多於corePoolSize，在有新任務新增時則選加入佇列，不直接建立執行緒

  當佇列滿時，在有新任務時就建立新執行緒

threadFactory:使用ThreadFactory建立新執行緒，預設使用defaultThreadFactory建立執行緒

handle:定義處理被拒絕任務的策略，預設使用ThreadPoolExecutor.AbortPolicy,任務被拒絕時將丟擲RejectExecutorException

3.Executors：提供了一系列靜態工廠方法用於建立各種執行緒池

   newFixedThreadPool:建立可重用且固定執行緒數的執行緒池，核心執行緒數和最大執行緒數是一致的，採用LinkedBlockingQueue作為阻塞等待佇列。如果執行緒池中的所有執行緒(核心執行緒數量)都處於活動狀態，此時再提交任務就在佇列中等待，直到有可用執行緒；如果執行緒池中的某個執行緒由於異常而結束時，執行緒池就會再補充一條新執行緒。

方法簽名：

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  //使用一個基於FIFO排序的阻塞佇列，在所有corePoolSize執行緒都忙時新任務將在佇列中等待
                                  new LinkedBlockingQueue<Runnable>());
}

    newSingleThreadExecutor:建立一個單執行緒的Executor，如果該執行緒因為異常而結束就新建一條執行緒來繼續執行後續的任務，由於LinkedBlockingQueue是FIFO的，所以能夠保證一定按照順序完成任務。

原文是這麼說的：使用一個工作執行緒操作一個無界佇列。(但是要注意，如果這個執行緒在關閉之前的執行過程中出現故障而終止，如果需要執行後續任務，則會有一個新的執行緒代替它)任務被保證按順序執行，並且在任何給定的時間內不會有超過一個任務處於活動狀態。與其他等效的{@code newFixedThreadPool(1)}不同的是，返回的執行器保證不能重新配置以使用其他執行緒。

方法簽名：
public static ExecutorService newSingleThreadExecutor() {
   return new FinalizableDelegatedExecutorService
                     //corePoolSize和maximumPoolSize都等於，表示固定執行緒池大小為1
                        (new ThreadPoolExecutor(1, 1,
                                                0L, TimeUnit.MILLISECONDS,
                                                new LinkedBlockingQueue<Runnable>()));
}

   newScheduledThreadPool:建立一個可延遲執行或定期執行的執行緒池,採用的是DelayedWorkQueue延遲佇列。

方法簽名：

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory);
}

例1：（使用newScheduledThreadPool來模擬心跳機制）

public class HeartBeat {
     public static void main(String[] args) {
         ScheduledExecutorService executor = Executors.newScheduledThreadPool(5);
         Runnable task = new Runnable() {
             public void run() {
                 System.out.println("HeartBeat.........................");
             }
        };
         executor.scheduleAtFixedRate(task,5,3, TimeUnit.SECONDS);   //5秒後第一次執行，之後每隔3秒執行一次
     }
}

輸出：

HeartBeat....................... //5秒後第一次輸出
HeartBeat....................... //每隔3秒輸出一個

   newCachedThreadPool:建立可快取的執行緒池，如果執行緒池中的執行緒在60秒未被使用就將被移除，在執行新的任務時，當執行緒池中有之前建立的可用執行緒就重用可用執行緒，否則就新建一條執行緒。同時SynchronousQueue佇列是同步移交，並不會因為任務多就進行等待。

方法簽名：

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  //使用同步佇列，將任務直接提交給執行緒
                                  new SynchronousQueue<Runnable>());
}

例2：

public class ThreadPoolTest {

    public static void main(String[] args) throws InterruptedException {

     ExecutorService threadPool = Executors.newCachedThreadPool();//執行緒池裡面的執行緒數會動態變化，並可線上程被移除前重用

        for (int i = 1; i <= 3; i ++) {

            final  int task = i;   //10個任務

            //TimeUnit.SECONDS.sleep(1);

            threadPool.execute(new Runnable() {    //接受一個Runnable例項

                public void run() {

                        System.out.println("執行緒名字： " + Thread.currentThread().getName() +  "  任務名為： "+task);

                }

            });

        }

    }

}

 

輸出：（為每個任務新建一條執行緒，共建立了3條執行緒）

執行緒名字： pool-1-thread-1 任務名為： 1
執行緒名字： pool-1-thread-2 任務名為： 2
執行緒名字： pool-1-thread-3 任務名為： 3

去掉第6行的註釋其輸出如下：（始終重複利用一條執行緒，因為newCachedThreadPool能重用可用執行緒）

執行緒名字： pool-1-thread-1 任務名為： 1
執行緒名字： pool-1-thread-1 任務名為： 2
執行緒名字： pool-1-thread-1 任務名為： 3

通過使用Executor可以很輕易的實現各種調優  管理  監視  記錄日誌和錯誤報告等待。

newWorkStealingPool：建立一個執行緒池，該執行緒池維護足夠的執行緒以支援給定的並行級別，底層使用的是支援任務分解的執行緒池ForkJoinPool，並可以使用多個佇列來減少爭用。並行級別對應於積極參與或可用來參與任務處理的執行緒的最大數量。執行緒的實際數量可以動態地增加和減少並保證提交的任務執行的順序。並行級別決定了同一時刻最多有多少個執行緒在執行，如不傳並行級別引數，將預設為當前系統的CPU個數。

方法簽名：

public static ExecutorService newWorkStealingPool(int parallelism) {
    return new ForkJoinPool
        (parallelism,
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}

new ForkJoinPool：是jdk1.8引入，是一種支援任務分解的執行緒池，當提交給他的任務“過大”，他就會按照預先定義的規則將大任務分解成小任務，多執行緒併發執行。 一般要配合可分解任務介面ForkJoinTask來使用，ForkJoinTask有兩個實現它的抽象類：RecursiveAction和RecursiveTask，其區別是前者沒有返回值，後者有返回值。
方法簽名：

public ForkJoinPool() {
    this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
         defaultForkJoinWorkerThreadFactory, null, false);
}

例3：

public static void main(String[] args)throws InterruptedException{
    ForkJoinPool pool =  new ForkJoinPool();
    Task task = new Task(60);
        pool.submit(task);
    pool.awaitTermination(6000,TimeUnit.MILLISECONDS);
    pool.shutdown();

}

public static class Task extends RecursiveAction{

    private static final long serialVersionUID = 1L;
    //定義一個分解任務的閾值——50,即一個任務最多承擔50個工作量
    int THRESHOLD=50;
    //任務量
    int task_Num=0;
    Task(int Num){
        this.task_Num=Num;
    }
    
    @Override
    protected void compute() {
        if(task_Num<=THRESHOLD){
            System.out.println(Thread.currentThread().getName()+"承擔了"+task_Num+"份工作");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }else{
            //隨機解成兩個任務
            Random m=new Random();
            int x=m.nextInt(50);

            Task left=new Task(x);
            Task right=new Task(task_Num-x);

            left.fork();
            right.fork();
        }
    }
}

 

4.Executor的生命週期

ExecutorService提供了管理Eecutor生命週期的方法，ExecutorService的生命週期包括了：執行  關閉和終止三種狀態。

ExecutorService在初始化建立時處於執行狀態。

shutdown方法等待提交的任務執行完成並不再接受新任務，在完成全部提交的任務後關閉

shutdownNow方法將強制終止所有執行中的任務並不再允許提交新任務

可以將一個Runnable（如例2）或Callable（如例3）提交給ExecutorService的submit方法執行，最終返回一上Futire用來獲得任務的執行結果或取消任務

例3：（任務執行完成後並返回執行結果）

public class CallableAndFuture {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ExecutorService executor = Executors.newSingleThreadExecutor();

        Future<String> future = executor.submit(new Callable<String>() {   //接受一上callable例項

            public String call() throws Exception {

                return "MOBIN";

            }

        });

        System.out.println("任務的執行結果："+future.get());

    }

}

 

輸出：

任務的執行結果：MOBIN

ExecutorCompletionService:實現了CompletionService，將執行完成的任務放到阻塞佇列中，通過take或poll方法來獲得執行結果

例4：（啟動10條執行緒，誰先執行完成就返回誰）

public class CompletionServiceTest {

    public static void main(String[] args) throws InterruptedException, ExecutionException {

        ExecutorService executor = Executors.newFixedThreadPool(10);        //建立含10.條執行緒的執行緒池

        CompletionService completionService = new ExecutorCompletionService(executor);

        for (int i =1; i <=10; i ++) {

            final  int result = i;

            completionService.submit(new Callable() {

                public Object call() throws Exception {

                    Thread.sleep(new Random().nextInt(5000));   //讓當前執行緒隨機休眠一段時間

                    return result;

                }

            });

        }

        System.out.println(completionService.take().get());   //獲取執行結果

    }

}

 

輸出結果可能每次都不同（在1到10之間）

3

通過Executor來設計應用程式可以簡化開發過程，提高開發效率，並有助於實現併發，在開發中如果需要建立執行緒可優先考慮使用Executor

相關推薦

java併發程式設計--Executor框架以及Executors類的建立執行緒池方法

       Eexecutor作為靈活且強大的非同步執行框架，其支援多種不同型別的任務執行策略，提供了一種標準的方法將任務的提交過程和執行過程解耦開發，基於生產者-消費者模式，其提交任務的執行緒相當於生產者，執行任務的執行緒相當於消費者，並用Runnab

Java併發程式設計的藝術筆記（八）——執行緒池

一.執行緒池的主要處理流程   ThreadPoolExecutor執行execute方法分下面4種情況。 1）如果當前執行的執行緒少於corePoolSize，則建立新執行緒來執行任務（注意，執行這一步需要獲取全域性鎖）。 2）如果執行的執行緒等於或多於corePoolSize，則將任

java併發程式設計--Executor框架

摘要： Eexecutor作為靈活且強大的非同步執行框架，其支援多種不同型別的任務執行策略，提供了一種標準的方法將任務的提交過程和執行過程解耦開發，基於生產者-消費者模式，其提交任務的執行緒相當於生產者，執行任務的執行緒相當於消費者，並用Runnable

Java併發程式設計-Executor框架之Callable和Future介面

在上一篇文章中我們已經瞭解了Executor框架進行執行緒管理，這篇文章將學習Executor框架的另一個特性，我們知道執行Runnable任務是沒有返回值得，但Executor可以執行併發任務並

Java Executor併發框架（四）建立執行緒池的核心引數的解釋

private final BlockingQueue<Runnable> workQueue; // 任務阻塞佇列 private final ReentrantLock mainLock = new ReentrantLock(); // 互斥鎖 privat

java併發程式設計之使用 CountDownLatch 控制多個執行緒執行順序

有時候會有這樣的需求，多個執行緒同時工作，然後其中幾個可以隨意併發執行，但有一個執行緒需要等其他執行緒工作結束後，才能開始。舉個例子，開啟多個執行緒分塊下載一個大檔案，每個執行緒只下載固定的一截，最後由另外一個執行緒來拼接所有的分段，那麼這時候我們可以考慮使用CountDownLatch來控制併發。

【搞定Java併發程式設計】第4篇：多執行緒概述~下篇

上一篇：多執行緒上篇：https://blog.csdn.net/pcwl1206/article/details/84837530 目  錄： 1、等待/喚醒機制 2、執行緒中斷 3、執行緒終止 4、執行緒休眠sleep 5、執行緒讓步yield() 6、jo

【搞定Java併發程式設計】第3篇：多執行緒概述~上篇

上一篇：併發基礎概述：https://blog.csdn.net/pcwl1206/article/details/84833911 目  錄： 1、什麼是執行緒 2、執行緒的建立 2.1、Thread和Runnable簡介 2.2、Thread和Runnable的異同

Java 併發程式設計系列之帶你瞭解多執行緒

早期的計算機不包含作業系統，它們從頭到尾執行一個程式，這個程式可以訪問計算機中的所有資源。在這種情況下，每次都只能執行一個程式，對於昂貴的計算機資源來說是一種嚴重的浪費。 作業系統出現後，計算機可以執行多個程式，不同的程式在單獨的程序中執行。作業系統負責為各個獨

java 併發程式設計學習筆記（九）多執行緒併發拓展

                                         多執行緒

淺談Java併發程式設計系列（一）—— 如何保證執行緒安全

執行緒安全類 保證類執行緒安全的措施： 不共享執行緒間的變數； 設定屬性變數為不可變變數； 每個共享的可變變數都使用一個確定的鎖保護； 保證執行緒安全的思路： 通過架構設計 通過

Java 併發程式設計(四)：如何保證物件的執行緒安全性

01、前言 先讓我吐一句肺腑之言吧，不說出來會憋出內傷的。《Java 併發程式設計實戰》這本書太特麼枯燥了，儘管它被奉為併發程式設計當中的經典之作，但我還是忍不住。因為第四章“物件的組合”我整整啃了兩週的時間，才啃出來點肉絲。 讀者朋友們見諒啊。要怪只能怪我自己的學習能力有限，真讀不了這種生硬無趣的技術書。但

使用Executors工廠類建立執行緒池

介面Executor僅僅是一種規範，是一種宣告，是一種定義，並沒有實現任何的功能，所以大多數的情況下，需要使用介面的實現類來完成指定的功能，比如ThreadPoolExecutor類就是Executor的實現類，但ThreadPoolExecutor在使用上並不是那麼方便，

【Java併發學習四】如何實現一個定時執行緒池

所謂 “定時任務執行緒池” 就是指放入執行緒池的任務，可以按照指定的等待週期迴圈執行。 Java裡面ScheduledThreadPoolExecutor這個類實現了這種功能。Spring裡面的定時任務也是在ScheduledThreadPoolExecu

Java併發庫（九、十）：執行緒池、Callable、Future

深切懷念傳智播客張孝祥老師，特將其代表作——Java併發庫視訊研讀兩遍，受益頗豐，記以後閱 http://blog.csdn.net/mmc_maodun/article/category/1773509 09.java5執行緒併發庫的應用        如果沒有執行緒池，

為什麼不推薦通過Executors直接建立執行緒池

通過Executors的方法創建出來的執行緒池都實現了ExecutorSerivice介面。常用的方法有 newFixedThreadPool(int Threads):建立固定數目的執行緒池。 newSingleThreadPoolExecutor():建立一個單執行緒化的Executor n

【搞定面試官】你還在用Executors來建立執行緒池？會有什麼問題呢？

前言 上文我們介紹了JDK中的執行緒池框架Executor。我們知道，只要需要建立執行緒的情況下，即使是在單執行緒模式下，我們也要儘量使用Executor。即： ExecutorService fixedThreadPool = Executors.newFixedThreadPool(1); //此處不該

Java併發程式設計 - Executor，Executors，ExecutorService, CompletionServie,Future,Callable

一、Exectuor框架簡介        Java從1.5版本開始，為簡化多執行緒併發程式設計，引入全新的併發程式設計包:java.util.concurrent及其併發程式設計框架（Executor框架）。 Executor框架是指ja

【Java併發】Executor框架

Executor框架 Executor框架結構 ThreadPoolExecutor FixedThreadPool SingleThreadExecutor newCachedThreadPool Schedul

Java併發程式設計3 —— 物件鎖和類鎖

synchronized關鍵字作用在同步程式碼塊上，給共享變數“上鎖”可以解決執行緒安全問題。這把“鎖”可以作用在某個物件上，也可以作用在某個類上。 舉個栗子，有個自助銀行，裡面有兩臺ATM機，工作人員可以看到每次存取款之後機器裡鈔票的總金額數。現在有兩個人來存錢，各存50