天天看點
傳回

[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 個線程，并交替執行。
    [Java并發程式設計（一）] 線程池 FixedThreadPool vs CachedThreadPool ...
    當 sleep 間隔為 10 milliseconds 時，共建立了 2 個線程，交替執行。
    [Java并發程式設計（一）] 線程池 FixedThreadPool vs CachedThreadPool ...
  • 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 個線程，并交替執行。
    [Java并發程式設計（一）] 線程池 FixedThreadPool vs CachedThreadPool ...
  • 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 個線程。
    [Java并發程式設計（一）] 線程池 FixedThreadPool vs 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問題的一些想法

上一篇: VMware伺服器虛拟化平台應急方案
下一篇: Paxos 實作日志複制同步（Basic Paxos）