從 Java 5 開(kāi)始，Java 提供了自己的線(xiàn)程池。線(xiàn)程池就是一個(gè)線(xiàn)程的容器，每次只執(zhí)行額定數(shù)量的線(xiàn)程。 java.util.concurrent.ThreadPoolExecutor 就是這樣的線(xiàn)程池。它很靈活，但使用起來(lái)也比較復(fù)雜，本文就對(duì)其做一個(gè)介紹。

首先是構(gòu)造函數(shù)。以最簡(jiǎn)單的構(gòu)造函數(shù)為例：

public ThreadPoolExecutor( 
      int corePoolSize, 
      int maximumPoolSize, 
      long keepAliveTime, 
      TimeUnit unit, 
      BlockingQueue<Runnable> workQueue) 

看起來(lái)挺復(fù)雜的。這里介紹一下。
corePoolSize 指的是保留的線(xiàn)程池大小。
maximumPoolSize 指的是線(xiàn)程池的最大大小。
keepAliveTime 指的是空閑線(xiàn)程結(jié)束的超時(shí)時(shí)間。
unit 是一個(gè)枚舉，表示 keepAliveTime 的單位。
workQueue 表示存放任務(wù)的隊(duì)列。
我們可以從線(xiàn)程池的工作過(guò)程中了解這些參數(shù)的意義。線(xiàn)程池的工作過(guò)程如下：

1、線(xiàn)程池剛創(chuàng)建時(shí)，里面沒(méi)有一個(gè)線(xiàn)程。任務(wù)隊(duì)列是作為參數(shù)傳進(jìn)來(lái)的。不過(guò)，就算隊(duì)列里面有任務(wù)，線(xiàn)程池也不會(huì)馬上執(zhí)行它們。
 
2、當(dāng)調(diào)用 execute() 方法添加一個(gè)任務(wù)時(shí)，線(xiàn)程池會(huì)做如下判斷：
    a. 如果正在運(yùn)行的線(xiàn)程數(shù)量小于 corePoolSize，那么馬上創(chuàng)建線(xiàn)程運(yùn)行這個(gè)任務(wù)；
    b. 如果正在運(yùn)行的線(xiàn)程數(shù)量大于或等于 corePoolSize，那么將這個(gè)任務(wù)放入隊(duì)列。
    c. 如果這時(shí)候隊(duì)列滿(mǎn)了，而且正在運(yùn)行的線(xiàn)程數(shù)量小于 maximumPoolSize，那么還是要?jiǎng)?chuàng)建線(xiàn)程運(yùn)行這個(gè)任務(wù)；
    d. 如果隊(duì)列滿(mǎn)了，而且正在運(yùn)行的線(xiàn)程數(shù)量大于或等于 maximumPoolSize，那么線(xiàn)程池會(huì)拋出異常，告訴調(diào)用者“我不能再接受任務(wù)了”。
 
3、當(dāng)一個(gè)線(xiàn)程完成任務(wù)時(shí)，它會(huì)從隊(duì)列中取下一個(gè)任務(wù)來(lái)執(zhí)行。
 
4、當(dāng)一個(gè)線(xiàn)程無(wú)事可做，超過(guò)一定的時(shí)間（keepAliveTime）時(shí)，線(xiàn)程池會(huì)判斷，如果當(dāng)前運(yùn)行的線(xiàn)程數(shù)大于 corePoolSize，那么這個(gè)線(xiàn)程就被停掉。所以線(xiàn)程池的所有任務(wù)完成后，它最終會(huì)收縮到 corePoolSize 的大小。
 
這樣的過(guò)程說(shuō)明，并不是先加入任務(wù)就一定會(huì)先執(zhí)行。假設(shè)隊(duì)列大小為 10，corePoolSize 為 3，maximumPoolSize 為 6，那么當(dāng)加入 20 個(gè)任務(wù)時(shí)，執(zhí)行的順序就是這樣的：首先執(zhí)行任務(wù) 1、2、3，然后任務(wù) 4~13 被放入隊(duì)列。這時(shí)候隊(duì)列滿(mǎn)了，任務(wù) 14、15、16 會(huì)被馬上執(zhí)行，而任務(wù) 17~20 則會(huì)拋出異常。最終順序是：1、2、3、14、15、16、4、5、6、7、8、9、10、11、12、13。下面是一個(gè)線(xiàn)程池使用的例子：

public static void main(String[] args) { 
  BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(); 
  ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 6, 1, TimeUnit.DAYS, queue); 
  
  for (int i = 0; i < 20; i++) { 
    executor.execute(new Runnable() { 
  
      public void run() { 
        try { 
          Thread.sleep(1000); 
        } catch (InterruptedException e) { 
          e.printStackTrace(); 
        } 
        System.out.println(String.format("thread %d finished", this.hashCode())); 
      } 
    }); 
  } 
  executor.shutdown(); 
} 

對(duì)這個(gè)例子的說(shuō)明如下：
 
1、BlockingQueue 只是一個(gè)接口，常用的實(shí)現(xiàn)類(lèi)有 LinkedBlockingQueue 和 ArrayBlockingQueue。用 LinkedBlockingQueue 的好處在于沒(méi)有大小限制。這樣的話(huà)，因?yàn)殛?duì)列不會(huì)滿(mǎn)，所以 execute() 不會(huì)拋出異常，而線(xiàn)程池中運(yùn)行的線(xiàn)程數(shù)也永遠(yuǎn)不會(huì)超過(guò) corePoolSize 個(gè)，keepAliveTime 參數(shù)也就沒(méi)有意義了。
 
2、shutdown() 方法不會(huì)阻塞。調(diào)用 shutdown() 方法之后，主線(xiàn)程就馬上結(jié)束了，而線(xiàn)程池會(huì)繼續(xù)運(yùn)行直到所有任務(wù)執(zhí)行完才會(huì)停止。如果不調(diào)用 shutdown() 方法，那么線(xiàn)程池會(huì)一直保持下去，以便隨時(shí)添加新的任務(wù)。
 
到這里對(duì)于這個(gè)線(xiàn)程池還只是介紹了一小部分。ThreadPoolExecutor 具有很強(qiáng)的可擴(kuò)展性，不過(guò)擴(kuò)展它的前提是要熟悉它的工作方式。后面的文章將會(huì)介紹如何擴(kuò)展 ThreadPoolExecutor 類(lèi)。

ava.util.concurrent.ThreadPoolExecutor 類(lèi)提供了豐富的可擴(kuò)展性。你可以通過(guò)創(chuàng)建它的子類(lèi)來(lái)自定義它的行為。例如，我希望當(dāng)每個(gè)任務(wù)結(jié)束之后打印一條消息，但我又無(wú)法修改任務(wù)對(duì)象，那么我可以這樣寫(xiě)：

ThreadPoolExecutor executor = new ThreadPoolExecutor(size, maxSize, 1, TimeUnit.DAYS, queue) {
  @Override
  protected void afterExecute(Runnable r, Throwable t) {
    System.out.println("Task finished.");
  }
};

除了 afterExecute 方法之外，ThreadPoolExecutor 類(lèi)還有 beforeExecute() 和 terminated() 方法可以重寫(xiě)，分別是在任務(wù)執(zhí)行之前和整個(gè)線(xiàn)程池停止之后執(zhí)行。

除了可以添加任務(wù)執(zhí)行前后的動(dòng)作之外， ThreadPoolExecutor 還允許你自定義當(dāng)添加任務(wù)失敗后的執(zhí)行策略。你可以調(diào)用線(xiàn)程池的 setRejectedExecutionHandler() 方法，用自定義的 RejectedExecutionHandler 對(duì)象替換現(xiàn)有的策略。 ThreadPoolExecutor 提供 4 個(gè)現(xiàn)有的策略，分別是：
ThreadPoolExecutor.AbortPolicy：表示拒絕任務(wù)并拋出異常
ThreadPoolExecutor.DiscardPolicy：表示拒絕任務(wù)但不做任何動(dòng)作
ThreadPoolExecutor.CallerRunsPolicy：表示拒絕任務(wù)，并在調(diào)用者的線(xiàn)程中直接執(zhí)行該任務(wù)
ThreadPoolExecutor.DiscardOldestPolicy：表示先丟棄任務(wù)隊(duì)列中的第一個(gè)任務(wù)，然后把這個(gè)任務(wù)加進(jìn)隊(duì)列。
這里是一個(gè)例子：
ThreadPoolExecutor executor = new ThreadPoolExecutor(size, maxSize, 1, TimeUnit.DAYS, queue);
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());

除此之外，你也可以通過(guò)實(shí)現(xiàn) RejectedExecutionHandler 接口來(lái)編寫(xiě)自己的策略。下面是一個(gè)例子：

ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 6, 1, TimeUnit.SECONDS, queue,
    new RejectedExecutionHandler() {
      public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println(String.format("Task %d rejected.", r.hashCode()));
      }
    }
);

最新評(píng)論