ThreadPoolExecutor使用詳解

ThreadPoolExecutor機制 

1、概述 
一、ThreadPoolExecutor做爲java.util.concurrent包對外提供基礎實現，之內部線程池的形式對外提供管理任務執行，線程調度，線程池管理等等服務； 
二、Executors方法提供的線程服務，都是經過參數設置來實現不一樣的線程池機制。 
三、先來了解其線程池管理的機制，有助於正確使用，避免錯誤使用致使嚴重故障。同時能夠根據本身的需求實現本身的線程池 

2、核心構造方法講解 
下面是ThreadPoolExecutor最核心的構造方法 

構造方法參數講解 

參數名	做用
corePoolSize	核心線程池大小
maximumPoolSize	最大線程池大小
keepAliveTime	線程池中超過corePoolSize數目的空閒線程最大存活時間；能夠allowCoreThreadTimeOut(true)使得核心線程有效時間
TimeUnit	keepAliveTime時間單位
workQueue	阻塞任務隊列
threadFactory	新建線程工廠
RejectedExecutionHandler	當提交任務數超過maxmumPoolSize+workQueue之和時，任務會交給RejectedExecutionHandler來處理

重點講解： 
其中比較容易讓人誤解的是：corePoolSize，maximumPoolSize，workQueue之間關係。 

1.當線程池小於corePoolSize時，新提交任務將建立一個新線程執行任務，即便此時線程池中存在空閒線程。 
2.當線程池達到corePoolSize時，新提交任務將被放入workQueue中，等待線程池中任務調度執行 
3.當workQueue已滿，且maximumPoolSize>corePoolSize時，新提交任務會建立新線程執行任務 
4.當提交任務數超過maximumPoolSize時，新提交任務由RejectedExecutionHandler處理 
5.當線程池中超過corePoolSize線程，空閒時間達到keepAliveTime時，關閉空閒線程 
6.當設置allowCoreThreadTimeOut(true)時，線程池中corePoolSize線程空閒時間達到keepAliveTime也將關閉 

線程管理機制圖示： 


3、Executors提供的線程池配置方案 

一、構造一個固定線程數目的線程池，配置的corePoolSize與maximumPoolSize大小相同，同時使用了一個無界LinkedBlockingQueue存放阻塞任務，所以多餘的任務將存在再阻塞隊列，不會由RejectedExecutionHandler處理 

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

二、構造一個緩衝功能的線程池，配置corePoolSize=0，maximumPoolSize=Integer.MAX_VALUE，keepAliveTime=60s,以及一個無容量的阻塞隊列 SynchronousQueue，所以任務提交以後，將會建立新的線程執行；線程空閒超過60s將會銷燬 

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

三、構造一個只支持一個線程的線程池，配置corePoolSize=maximumPoolSize=1，無界阻塞隊列LinkedBlockingQueue；保證任務由一個線程串行執行 

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

四、構造有定時功能的線程池，配置corePoolSize，無界延遲阻塞隊列DelayedWorkQueue；有意思的是：maximumPoolSize=Integer.MAX_VALUE，因爲DelayedWorkQueue是無界隊列，因此這個值是沒有意義的 

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

public static ScheduledExecutorService newScheduledThreadPool(
            int corePoolSize, ThreadFactory threadFactory) {
        return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
    }

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

4、定製屬於本身的非阻塞線程池 

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


public class CustomThreadPoolExecutor {

	
	private ThreadPoolExecutor pool = null;
	
	
	/**
	 * 線程池初始化方法
	 * 
	 * corePoolSize 核心線程池大小----10
	 * maximumPoolSize 最大線程池大小----30
	 * keepAliveTime 線程池中超過corePoolSize數目的空閒線程最大存活時間----30+單位TimeUnit
	 * TimeUnit keepAliveTime時間單位----TimeUnit.MINUTES
	 * workQueue 阻塞隊列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞隊列
	 * threadFactory 新建線程工廠----new CustomThreadFactory()====定製的線程工廠
	 * rejectedExecutionHandler 當提交任務數超過maxmumPoolSize+workQueue之和時,
	 * 							即當提交第41個任務時(前面線程都沒有執行完,此測試方法中用sleep(100)),
	 * 						          任務會交給RejectedExecutionHandler來處理
	 */
	public void init() {
		pool = new ThreadPoolExecutor(
				10,
				30,
				30,
				TimeUnit.MINUTES,
				new ArrayBlockingQueue<Runnable>(10),
				new CustomThreadFactory(),
				new CustomRejectedExecutionHandler());
	}

	
	public void destory() {
		if(pool != null) {
			pool.shutdownNow();
		}
	}
	
	
	public ExecutorService getCustomThreadPoolExecutor() {
		return this.pool;
	}
	
	private class CustomThreadFactory implements ThreadFactory {

		private AtomicInteger count = new AtomicInteger(0);
		
		@Override
		public Thread newThread(Runnable r) {
			Thread t = new Thread(r);
			String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);
			System.out.println(threadName);
			t.setName(threadName);
			return t;
		}
	}
	
	
	private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {

		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			// 記錄異常
			// 報警處理等
			System.out.println("error.............");
		}
	}
	
	
	
	// 測試構造的線程池
	public static void main(String[] args) {
		CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor();
		// 1.初始化
		exec.init();
		
		ExecutorService pool = exec.getCustomThreadPoolExecutor();
		for(int i=1; i<100; i++) {
			System.out.println("提交第" + i + "個任務!");
			pool.execute(new Runnable() {
				@Override
				public void run() {
					try {
						Thread.sleep(3000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("running=====");
				}
			});
		}
		
		
		
		// 2.銷燬----此處不能銷燬,由於任務沒有提交執行完,若是銷燬線程池,任務也就沒法執行了
		// exec.destory();
		
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

方法中創建一個核心線程數爲30個，緩衝隊列有10個的線程池。每一個線程任務，執行時會先睡眠3秒，保證提交10任務時，線程數目被佔用完，再提交30任務時，阻塞隊列被佔用完，，這樣提交第41個任務是，會交給CustomRejectedExecutionHandler 異常處理類來處理。 

提交任務的代碼以下： 

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         * Proceed in 3 steps:
         *
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

注意：41之後提交的任務就不能正常處理了，由於，execute中提交到任務隊列是用的offer方法，如上面代碼，這個方法是非阻塞的，因此就會交給CustomRejectedExecutionHandler 來處理，因此對於大數據量的任務來講，這種線程池，若是不設置隊列長度會OOM，設置隊列長度，會有任務得不處處理，接下來咱們構建一個阻塞的自定義線程池 

5、定製屬於本身的阻塞線程池 

package com.tongbanjie.trade.test.commons;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class CustomThreadPoolExecutor {  
	  
      
    private ThreadPoolExecutor pool = null;  
      
      
    /** 
     * 線程池初始化方法 
     *  
     * corePoolSize 核心線程池大小----1 
     * maximumPoolSize 最大線程池大小----3 
     * keepAliveTime 線程池中超過corePoolSize數目的空閒線程最大存活時間----30+單位TimeUnit 
     * TimeUnit keepAliveTime時間單位----TimeUnit.MINUTES 
     * workQueue 阻塞隊列----new ArrayBlockingQueue<Runnable>(5)====5容量的阻塞隊列 
     * threadFactory 新建線程工廠----new CustomThreadFactory()====定製的線程工廠 
     * rejectedExecutionHandler 當提交任務數超過maxmumPoolSize+workQueue之和時, 
     *                          即當提交第41個任務時(前面線程都沒有執行完,此測試方法中用sleep(100)), 
     *                                任務會交給RejectedExecutionHandler來處理 
     */  
    public void init() {  
        pool = new ThreadPoolExecutor(  
                1,  
                3,  
                30,  
                TimeUnit.MINUTES,  
                new ArrayBlockingQueue<Runnable>(5),  
                new CustomThreadFactory(),  
                new CustomRejectedExecutionHandler());  
    }  
  
      
    public void destory() {  
        if(pool != null) {  
            pool.shutdownNow();  
        }  
    }  
      
      
    public ExecutorService getCustomThreadPoolExecutor() {  
        return this.pool;  
    }  
      
    private class CustomThreadFactory implements ThreadFactory {  
  
        private AtomicInteger count = new AtomicInteger(0);  
          
        @Override  
        public Thread newThread(Runnable r) {  
            Thread t = new Thread(r);  
            String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);  
            System.out.println(threadName);  
            t.setName(threadName);  
            return t;  
        }  
    }  
      
      
    private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {  
  
        @Override  
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {  
        	try {
                                // 核心改造點，由blockingqueue的offer改爲put阻塞方法
				executor.getQueue().put(r);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
        }  
    }  
      
      
      
    // 測試構造的線程池  
    public static void main(String[] args) {  
    	
        CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor();  
        // 1.初始化  
        exec.init();  
          
        ExecutorService pool = exec.getCustomThreadPoolExecutor();  
        for(int i=1; i<100; i++) {  
            System.out.println("提交第" + i + "個任務!");  
            pool.execute(new Runnable() {  
                @Override  
                public void run() {  
                    try {  
                    	System.out.println(">>>task is running====="); 
                        TimeUnit.SECONDS.sleep(10);
                    } catch (InterruptedException e) {  
                        e.printStackTrace();  
                    }  
                }  
            });  
        }  
          
          
        // 2.銷燬----此處不能銷燬,由於任務沒有提交執行完,若是銷燬線程池,任務也就沒法執行了  
        // exec.destory();  
          
        try {  
            Thread.sleep(10000);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}  

解釋：當提交任務被拒絕時，進入拒絕機制，咱們實現拒絕方法，把任務從新用阻塞提交方法put提交，實現阻塞提交任務功能，防止隊列過大，OOM，提交被拒絕方法在下面 

   

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();

        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        else if (!addWorker(command, false))
            // 進入拒絕機制， 咱們把runnable任務拿出來，從新用阻塞操做put，來實現提交阻塞功能
            reject(command);
    }

總結： 
一、用ThreadPoolExecutor自定義線程池，看線程是的用途，若是任務量不大，能夠用無界隊列，若是任務量很是大，要用有界隊列，防止OOM 二、若是任務量很大，還要求每一個任務都處理成功，要對提交的任務進行阻塞提交，重寫拒絕機制，改成阻塞提交。保證不拋棄一個任務 三、最大線程數通常設爲2N+1最好，N是CPU核數 四、核心線程數，看應用，若是是任務，一天跑一次，設置爲0，合適，由於跑完就停掉了，若是是經常使用線程池，看任務量，是保留一個核心仍是幾個核心線程數 五、若是要獲取任務執行結果，用CompletionService，可是注意，獲取任務的結果的要從新開一個線程獲取，若是在主線程獲取，就要等任務都提交後才獲取，就會阻塞大量任務結果，隊列過大OOM，因此最好異步開個線程獲取結果