併發編程之線程池的使用及擴展和優化

時間  2019-11-30
標籤 併發 編程 線程 使用 擴展 優化 简体版
原文   原文鏈接

前言

多線程的軟件設計方法確實能夠最大限度的發揮現代多核處理器的計算能力，提升生產系統的吞吐量和性能。可是，若是一個系統同時建立大量線程，線程間頻繁的切換上下文致使的系統開銷將會拖慢整個系統。嚴重的甚至致使內存耗盡致使OOM異常。所以，在實際的生產環境中，線程的數量必須獲得控制，盲目的建立大量新車對系統是有傷害的。java

那麼，怎麼才能最大限度的利用CPU的性能，又能保持系統的穩定性呢？其中有一個方法就是使用線程池。程序員

簡而言之，在使用線程池後，建立線程便處理從線程池得到空閒線程，關閉線程變成了向池子歸還線程。也就是說，提升了線程的複用。算法

而 JDK 在 1.5 以後爲我提供了現成的線程池工具，咱們今天就來學習看看如何使用他們。緩存

  1. Executors 線程池工廠能建立哪些線程池
  2. 如何手動建立線程池
  3. 如何擴展線程池
  4. 如何優化線程池的異常信息
  5. 如何設計線程池中的線程數量

1. Executors 線程池工廠能建立哪些線程池

先來一個最簡單的線程池使用例子：多線程

static class MyTask implements Runnable {

    @Override
    public void run() {
      System.out
          .println(System.currentTimeMillis() + ": Thread ID :" + Thread.currentThread().getId());
      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  public static void main(String[] args) {
    MyTask myTask = new MyTask();
    ExecutorService service1 = Executors.newFixedThreadPool(5);
    for (int i = 0; i < 10; i++) {
      service1.submit(myTask);
    }
    service1.shutdown();
  }

複製代碼

運行結果：框架

運行結果

咱們建立了一個線程池實例，並設置默認線程數量爲5，並向線程池提交了10任務，分別打印當前毫秒時間和線程ID，從結果中，咱們能夠看到結果中有5個相同 id 的線程打印了毫秒時間。ide

這是最簡單的例子。工具

接下來咱們講講其餘的線程建立方式。性能

1. 固定線程池 ExecutorService service1 = Executors.newFixedThreadPool(5); 該方法返回一個固定線程數量的線程池。該線程池中的線程數量始終不變。當有一個新的任務提交時，線程池中如有空閒線程，則當即執行，若沒有，則新的任務會被暫存在一個任務隊列（默認無界隊列 int 最大數）中，待有線程空閒時，便處理在任務隊列中的任務。學習

2. 單例線程池 ExecutorService service3 = Executors.newSingleThreadExecutor(); 該方法返回一個只有一個線程的線程池。若多餘一個任務被提交到該線程池，任務會被保存在一個任務隊列（默認無界隊列 int 最大數）中，待線程空閒，按先入先出的順序執行隊列中的任務。

3. 緩存線程池 ExecutorService service2 = Executors.newCachedThreadPool(); 該方法返回一個可根據實際狀況調整線程數量的線程池，線程池的線程數量不肯定，但如有空閒線程能夠複用，則會優先使用可複用的線程，全部線程均在工做，若是有新的任務提交，則會建立新的線程處理任務。全部線程在當前任務執行完畢後，將返回線程池進行復用。

4. 任務調用線程池 ExecutorService service4 = Executors.newScheduledThreadPool(2); 該方法也返回一個 ScheduledThreadPoolExecutor 對象，該線程池能夠指定線程數量。

前3個線程的用法沒什麼差別，關鍵是第四個，雖然線程任務調度框架不少，可是咱們仍然能夠學習該線程池。如何使用呢？下面來個例子：

class A {

  public static void main(String[] args) {
    ScheduledThreadPoolExecutor service4 = (ScheduledThreadPoolExecutor) Executors
        .newScheduledThreadPool(2);

    // 若是前面的任務沒有完成，則調度也不會啓動
    service4.scheduleAtFixedRate(new Runnable() {
      @Override
      public void run() {
        try {
          // 若是任務執行時間大於間隔時間，那麼就以執行時間爲準（防止任務出現堆疊）。
          Thread.sleep(10000);
          System.out.println(System.currentTimeMillis() / 1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }// initialDelay（初始延遲） 表示第一次延時時間 ; period 表示間隔時間
    }, 0, 2, TimeUnit.SECONDS);


    service4.scheduleWithFixedDelay(new Runnable() {
      @Override
      public void run() {
        try {
          Thread.sleep(5000);
          System.out.println(System.currentTimeMillis() / 1000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }
      }// initialDelay（初始延遲） 表示延時時間；delay + 任務執行時間 = 等於間隔時間 period
    }, 0, 2, TimeUnit.SECONDS);

    // 在給定時間，對任務進行一次調度
    service4.schedule(new Runnable() {
      @Override
      public void run() {
        System.out.println("5 秒以後執行 schedule");
      }
    }, 5, TimeUnit.SECONDS);
  }
  }

}
複製代碼

上面的代碼建立了一個 ScheduledThreadPoolExecutor 任務調度線程池，分別調用了3個方法，須要着重解釋 scheduleAtFixedRate 和 scheduleWithFixedDelay 方法，這兩個方法的做用很類似，惟一的區別就是他們執行人物的間隔時間的計算方式，前者時間間隔算法是根據指定的 period 時間和任務執行時間中取時間長的，後者取的是指定的 delay 時間 + 任務執行時間。若是同窗們有興趣，能夠將上面的代碼跑跑看。同樣便能看出端倪。

好了，JDK 給咱們封裝了建立線程池的 4 個方法，可是，請注意，因爲這些方法高度封裝，所以，若是使用不當，出了問題將無從排查，所以，我建議，程序員應到本身手動建立線程池，而手動建立的前提就是高度瞭解線程池的參數設置。那麼咱們就來看看如何手動建立線程池。

2. 如何手動建立線程池

下面是一個手動建立線程池的範本：

/** * 默認5條線程（默認數量，即最少數量）， * 最大20線程（指定了線程池中的最大線程數量）， * 空閒時間0秒（當線程池梳理超過核心數量時，多餘的空閒時間的存活時間，即超過核心線程數量的空閒線程，在多長時間內，會被銷燬）， * 等待隊列長度1024， * 線程名稱[MXR-Task-%d],方便回溯， * 拒絕策略：當任務隊列已滿，拋出RejectedExecutionException * 異常。 */
  private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 20, 0L,
      TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024)
      , new ThreadFactoryBuilder().setNameFormat("My-Task-%d").build()
      , new AbortPolicy()
  );

複製代碼

咱們看到，ThreadPoolExecutor 也就是線程池有 7 個參數，咱們一塊兒來好好看看：

  1. corePoolSize 線程池中核心線程數量
  2. maximumPoolSize 最大線程數量
  3. keepAliveTime 空閒時間（當線程池梳理超過核心數量時，多餘的空閒時間的存活時間，即超過核心線程數量的空閒線程，在多長時間內，會被銷燬）
  4. unit 時間單位
  5. workQueue 當核心線程工做已滿，須要存儲任務的隊列
  6. threadFactory 建立線程的工廠
  7. handler 當隊列滿了以後的拒絕策略

前面幾個參數咱們就不講了，很簡單，主要是後面幾個參數，隊列，線程工廠，拒絕策略。

咱們先看看隊列，線程池默認提供了 4 個隊列。

  1. 無界隊列： 默認大小 int 最大值，所以可能會耗盡系統內存，引發OOM，很是危險。
  2. 直接提交的隊列 ： 沒有容量，不會保存，直接建立新的線程，所以須要設置很大的線程池數。不然容易執行拒絕策略，也很危險。
  3. 有界隊列：若是core滿了，則存儲在隊列中，若是core滿了且隊列滿了，則建立線程，直到maximumPoolSize 到了，若是隊列滿了且最大線程數已經到了，則執行拒絕策略。
  4. 優先級隊列：按照優先級執行任務。也能夠設置大小。

樓主在本身的項目中使用了無界隊列，可是設置了任務大小，1024。若是你的任務不少，建議分爲多個線程池。不要把雞蛋放在一個籃子裏。

再看看拒絕策略，什麼是拒絕策略呢？當隊列滿了，如何處理那些仍然提交的任務。JDK 默認有4種策略。

  1. AbortPolicy ：直接拋出異常，阻止系統正常工做.
  2. CallerRunsPolicy : 只要線程池未關閉，該策略直接在調用者線程中，運行當前被丟棄的任務。顯然這樣作不會真的丟棄任務，可是，任務提交線程的性能極有可能會急劇降低。
  3. DiscardOldestPolicy: 該策略將丟棄最老的一個請求，也就是即將被執行的一個任務，並嘗試再次提交當前任務.
  4. DiscardPolicy: 該策略默默地丟棄沒法處理的任務，不予任何處理，若是容許任務丟失，我以爲這是最好的方案.

固然，若是你不滿意JDK提供的拒絕策略，能夠本身實現，只須要實現 RejectedExecutionHandler 接口，並重寫 rejectedExecution 方法便可。

最後，線程工廠，線程池的全部線程都由線程工廠來建立，而默認的線程工廠太過單一，咱們看看默認的線程工廠是如何建立線程的：

/** * The default thread factory */
    static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }
複製代碼

能夠看到，線程名稱爲 pool- + 線程池編號 + -thread- + 線程編號 。設置爲非守護線程。優先級爲默認。

若是咱們想修更名稱呢？對，實現 ThreadFactory 接口，重寫 newThread 方法便可。可是已經有人造好輪子了， 好比咱們的例子中使用的 google 的 guaua 提供的 ThreadFactoryBuilder 工廠。能夠自定義線程名稱，是否守護，優先級，異常處理等等，功能強大。

3. 如何擴展線程池

那麼咱們能擴展線程池的功能嗎？好比記錄線程任務的執行時間。實際上，JDK 的線程池已經爲咱們預留的接口，在線程池核心方法中，有2 個方法是空的，就是給咱們預留的。還有一個線程池退出時會調用的方法。咱們看看例子：

/** * 如何擴展線程池，重寫 beforeExecute, afterExecute, terminated 方法，這三個方法默認是空的。 * * 能夠監控每一個線程任務執行的開始和結束時間，或者自定義一些加強。 * * 在 Worker 的 runWork 方法中，會調用這些方法 */
public class ExtendThreadPoolDemo {

  static class MyTask implements Runnable {

    String name;

    public MyTask(String name) {
      this.name = name;
    }

    @Override
    public void run() {
      System.out
          .println("正在執行：Thread ID:" + Thread.currentThread().getId() + ", Task Name = " + name);
      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }


  public static void main(String[] args) throws InterruptedException {
    ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<>()) {
      @Override
      protected void beforeExecute(Thread t, Runnable r) {
        System.out.println("準備執行：" + ((MyTask) r).name);
      }

      @Override
      protected void afterExecute(Runnable r, Throwable t) {
        System.out.println("執行完成： " + ((MyTask) r).name);
      }

      @Override
      protected void terminated() {
        System.out.println("線程池退出");
      }
    };

    for (int i = 0; i < 5; i++) {
      MyTask myTask = new MyTask("TASK-GEYM-" + i);
      es.execute(myTask);
      Thread.sleep(10);

    }

    es.shutdown();
  }

}

複製代碼

咱們重寫了 beforeExecute 方法，也就是執行任務以前會調用該方法，而 afterExecute 方法則是在任務執行完畢後會調用該方法。還有一個 terminated 方法，在線程池退出時會調用該方法。執行結果是什麼呢？

能夠看到，每一個任務執行先後都會調用 before 和 after 方法。至關於執行了一個切面。而在調用 shutdown 方法後則會調用 terminated 方法。

4. 如何優化線程池的異常信息

如何優化線程池的異常信息？ 在說這個問題以前，咱們先說一個不容易發現的bug：

看代碼：

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

    ThreadPoolExecutor executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 0L,
        TimeUnit.MILLISECONDS, new SynchronousQueue<>());

    for (int i = 0; i < 5; i++) {
      executor.submit(new DivTask(100, i));
    }


  }


  static class DivTask implements Runnable {
    int a, b;

    public DivTask(int a, int b) {
      this.a = a;
      this.b = b;
    }

    @Override
    public void run() {
      double re = a / b;
      System.out.println(re);
    }
  }
複製代碼

執行結果：

注意：只有4個結果，其中一個結果被吞沒了，而且沒有任何信息。爲何呢？若是仔細看代碼，會發現，在進行 100 / 0 的時候確定會報錯的，可是卻沒有報錯信息，使人頭痛，爲何呢？實際上，若是你使用 execute 方法則會打印錯誤信息，當你使用 submit 方法卻沒有調用它的get 方法，異常將會被吞沒，由於，若是發生了異常，異常是做爲返回值返回的。

怎麼辦呢？咱們固然可使用 execute 方法，可是咱們能夠有另外一種方式：重寫 submit 方法，樓主寫了一個例子，你們看一下：

static class TraceThreadPoolExecutor extends ThreadPoolExecutor {

    public TraceThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
      super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    @Override
    public void execute(Runnable command) {
// super.execute(command);
      super.execute(wrap(command, clientTrace(), Thread.currentThread().getName()));
    }

    @Override
    public Future<?> submit(Runnable task) {
// return super.submit(task);
      return super.submit(wrap(task, clientTrace(), Thread.currentThread().getName()));
    }

    private Exception clientTrace() {
      return new Exception("Client stack trace");
    }


    private Runnable wrap(final Runnable task, final Exception clientStack, String clientThreaName) {
      return new Runnable() {
        @Override
        public void run() {
          try {
            task.run();
          } catch (Exception e) {
            e.printStackTrace();
            clientStack.printStackTrace();
            throw e;
          }
        }
      };
    }
  }
複製代碼

咱們重寫了 submit 方法，封裝了異常信息，若是發生了異常，將會打印堆棧信息。咱們看看使用重寫後的線程池後的結果是什麼？

從結果中，咱們清楚的看到了錯誤信息的緣由：by zero！而且堆棧信息明確，方便排錯。優化了默認線程池的策略。

5. 如何設計線程池中的線程數量

線程池的大小對系統的性能有必定的影響，過大或者太小的線程數量都沒法發揮最優的系統性能，可是線程池大小的肯定也不須要作的很是精確。由於只要避免極大和極小兩種狀況，線程池的大小對性能的影響都不會影響太大，通常來講，肯定線程池的大小須要考慮CPU數量，內存大小等因素，在《Java Concurrency in Practice》 書中給出了一個估算線程池大小的經驗公式：

公式仍是有點複雜的，簡單來講，就是若是你是CPU密集型運算，那麼線程數量和CPU核心數相同就好，避免了大量無用的切換線程上下文，若是你是IO密集型的話，須要大量等待，那麼線程數能夠設置的多一些，好比CPU核心乘以2.

至於如何獲取 CPU 核心數，Java 提供了一個方法：

Runtime.getRuntime().availableProcessors()；

返回了CPU的核心數量。

總結

好了，到這裏，咱們已經對如何使用線程池有了一個認識，這裏，樓主建議你們手動建立線程池，這樣對線程池中的各個參數能夠有精準的瞭解，在對系統進行排錯或者調優的時候有好處。好比設置核心線程數多少合適，最大線程數，拒絕策略，線程工廠，隊列的大小和類型等等，也能夠是G家的線程工廠自定義線程。

下一篇，咱們將深刻源碼，看看JDK 的線程池是如何實現的。所以，先熟悉線程池的使用吧！！！

good luck！！！

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程