[Java併發編程（一）] 線程池 FixedThreadPool vs CachedThreadPool ...

[Java併發編程（一）] 線程池 FixedThreadPool vs CachedThreadPool ...

摘要

介紹 Java 併發包裏的幾個主要 ExecutorService 。

正文

CachedThreadPool

CachedThreadPool 是經過 java.util.concurrent.Executors 建立的 ThreadPoolExecutor 實例。這個實例會根據須要，在線程可用時，重用以前構造好的池中線程。這個線程池在執行 大量短生命週期的異步任務時（many short-lived asynchronous task），能夠顯著提升程序性能。調用 execute 時，能夠重用以前已構造的可用線程，若是不存在可用線程，那麼會從新建立一個新的線程並將其加入到線程池中。若是線程超過 60 秒還未被使用，就會被停止並從緩存中移除。所以，線程池在長時間空閒後不會消耗任何資源。

注意隊列實例是：new SynchronousQueue ()

/**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to <tt>execute</tt> will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

FixedThreadPool

FixedThreadPool 是經過 java.util.concurrent.Executors 建立的 ThreadPoolExecutor 實例。這個實例會複用 固定數量的線程 處理一個 共享的無邊界隊列 。任什麼時候間點，最多有 nThreads 個線程會處於活動狀態執行任務。若是當全部線程都是活動時，有多的任務被提交過來，那麼它會一致在隊列中等待直到有線程可用。若是任何線程在執行過程當中由於錯誤而停止，新的線程會替代它的位置來執行後續的任務。全部線程都會一致存於線程池中，直到顯式的執行 ExecutorService.shutdown() 關閉。

注意隊列實例是：new LinkedBlockingQueue ()

/**
     * Creates a thread pool that reuses a fixed number of threads
     * operating off a shared unbounded queue.  At any point, at most
     * <tt>nThreads</tt> threads will be active processing tasks.
     * If additional tasks are submitted when all threads are active,
     * they will wait in the queue until a thread is available.
     * If any thread terminates due to a failure during execution
     * prior to shutdown, a new one will take its place if needed to
     * execute subsequent tasks.  The threads in the pool will exist
     * until it is explicitly {@link ExecutorService#shutdown shutdown}.
     *
     * @param nThreads the number of threads in the pool
     * @return the newly created thread pool
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */
    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

SingleThreadPool

SingleThreadPool 是經過 java.util.concurrent.Executors 建立的 ThreadPoolExecutor 實例。這個實例只會使用單個工做線程來執行一個無邊界的隊列。（注意，若是單個線程在執行過程當中由於某些錯誤停止，新的線程會替代它執行後續線程）。它能夠保證認爲是按順序執行的，任什麼時候候都不會有多於一個的任務處於活動狀態。和 newFixedThreadPool(1) 的區別在於，若是線程遇到錯誤停止，它是沒法使用替代線程的。

/**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue. (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time. Unlike the otherwise equivalent
     * <tt>newFixedThreadPool(1)</tt> the returned executor is
     * guaranteed not to be reconfigurable to use additional threads.
     *
     * @return the newly created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

程序演示

• LiftOff

  public class LiftOff implements Runnable {
          protected int countDown = 10; // Default
          private static int taskCount = 0;
          private final int id = taskCount++;
  
          public LiftOff() {}
  
          public LiftOff(int countDown) {
              this.countDown = countDown;
          }
  
          public String status() {
              return "Thread ID: [" + String.format("%3d", Thread.currentThread().getId()) + "] #" + id + "(" + (countDown > 0 ? countDown : "LiftOff!") + ") ";
          }
  
          @Override
          public void run() {
              while (countDown-- > 0) {
                  System.out.println(status());
                  Thread.yield();
              }
          }
  
      }

• CachedThreadPoolCase

  public class CachedThreadPoolCase {
          public static void main(String[] args) throws InterruptedException {
              ExecutorService exec = Executors.newCachedThreadPool();
              for(int i = 0; i < 5; i++) {
                  exec.execute(new LiftOff());
                  Thread.sleep(10);
              }
              exec.shutdown();
          }
      }

  當 sleep 間隔爲 5 milliseconds 時，共建立了 3 個線程，並交替執行。

  當 sleep 間隔爲 10 milliseconds 時，共建立了 2 個線程，交替執行。

• FixedThreadPoolCase

  public class FixedThreadPoolCase {
  
          public static void main(String[] args) throws InterruptedException {
              ExecutorService exec = Executors.newFixedThreadPool(3);
              for (int i = 0; i < 5; i++) {
                  exec.execute(new LiftOff());
                  Thread.sleep(10);
              }
              exec.shutdown();
          }
      }

  不管 sleep 間隔時間是多少，總共都建立 3 個線程，並交替執行。

• SingleThreadCase

  public class SingleThreadPoolCase {
  
          public static void main(String[] args) throws InterruptedException {
              ExecutorService exec = Executors.newSingleThreadExecutor();
              for (int i = 0; i < 10; i++) {
                  exec.execute(new LiftOff());
                  Thread.sleep(5);
              }
              exec.shutdown();
          }
      }

  不管 sleep 間隔時間是多少，總共都只建立 1 個線程。

FixedThreadPool 與 CachedThreadPool 特性對比

特性	FixedThreadPool	CachedThreadPool
重用	FixedThreadPool 與 CacheThreadPool差很少，也是能 reuse 就用，但不能隨時建新的線程	緩存型池子，先查看池中有沒有之前創建的線程，若是有，就 reuse ；若是沒有，就建一個新的線程加入池中
池大小	可指定 nThreads，固定數量	可增加，最大值 Integer.MAX_VALUE
隊列大小	無限制	無限制
超時	無 IDLE	默認 60 秒 IDLE
使用場景	因此 FixedThreadPool 多數針對一些很穩定很固定的正規併發線程，多用於服務器	大量短生命週期的異步任務
結束	不會自動銷燬	注意，放入 CachedThreadPool 的線程沒必要擔憂其結束，超過 TIMEOUT 不活動，其會自動被終止。

最佳實踐

FixedThreadPool 和 CachedThreadPool 二者對高負載的應用都不是特別友好。

CachedThreadPool 要比 FixedThreadPool 危險不少。

若是應用要求高負載、低延遲，最好不要選擇以上兩種線程池：

1. 任務隊列的無邊界：會致使內存溢出以及高延遲
2. 長時間運行會致使 CachedThreadPool 在線程建立上失控

由於二者都不是特別友好，因此推薦使用 ThreadPoolExecutor ，它提供了不少參數能夠進行細粒度的控制。

1. 將任務隊列設置成有邊界的隊列
2. 使用合適的 RejectionHandler - 自定義的 RejectionHandler 或 JDK 提供的默認 handler 。

3. 若是在任務完成先後須要執行某些操做，能夠重載

   beforeExecute(Thread, Runnable)
    afterExecute(Runnable, Throwable)

4. 重載 ThreadFactory ，若是有線程定製化的需求

5. 在運行時動態控制線程池的大小（Dynamic Thread Pool）

參考

iteye: Java 併發包中的幾種 ExecutorService

stackoverflow: FixedThreadPool vs CachedThreadPool: the lesser of two evils

blogjava: 深刻淺出多線程（4）對CachedThreadPool OutOfMemoryError問題的一些想法

結束