标签：
1 public ThreadPoolExecutor(int corePoolSize, 2 int maximumPoolSize, 3 long keepAliveTime, 4 TimeUnit unit, 5 BlockingQueue<Runnable> workQueue, 6 ThreadFactory threadFactory, 7 RejectedExecutionHandler handler)
corePoolSize：线程池核心线程个数。线程池在创建后默认情况下线程个数为0，当有新的任务到来时，如果池中当前线程个数小于corePoolSize，则创建一个新的线程。（tips：如果希望在线程池初始化时启动线程，可以调用prestartCoreThread()或者prestartAllCoreThreads()方法启动一个线程或者所有核心线程。）
maximumPoolSize：线程池最大线程个数
keepAliveTime：线程空闲时最大存活时间，默认情况下，keepAliveTime只对maximumPoolSize这部分的线程有效，如果希望keepAliveTime对corePoolSize线程也有效，可以调用allowCoreThreadTimeOut(boolean)方法。
unit：keepAliveTime的单位
workQueue：任务队列，这是一个阻塞队列（等价于非阻塞队列通过sychronized和wait/notify集成了同步以及empty/full判断的功能），用来保存来不及处理的任务。
threadFactory：线程创建工厂，线程池通过调用它的newThread方法创建新的线程，可以定义自己的线程创建工厂实现ThreadFactory接口。默认的ThreadFactory的newThread方法会给池中的线程命名。还记得我们在log日志中记录的当前线程名字吗？如果你的程序中使用了线程池，那么日志上打印的线程名字就是在这个方法里命名的。
handler：拒绝策略，当线程池的任务队列满了且线程个数达到最大线程个数的时候，需要对新到达的任务按拒绝策略处理。线程池中已经提供的拒绝策略有：
AbortPolicy——丢弃该任务，并抛出 RejectedExecutionException异常
DiscardPolicy——直接丢弃该任务
DiscardOldestPolicy——丢弃任务队列中最旧的任务，并尝试重新执行execute方法
CallerRunsPolicy——有调用线程执行该任务
默认情况下，线程池采用AbortPolicy，可以更改线程池的拒绝策略，也可以通过实现RejectedExecutionHandler接口定义自己的的拒绝策略。
我们先来直观了解一个任务被提交到线程池之后，他的执行流程：
通过调用ThreadPoolExecutor类中execute()方法可以提交任务到线程池执行，另外submit()方法也可以向线程池提交任务，两者的区别在于execute()方法不关心任务的执行结果，而submit()方法可以得到任务的执行结果，并且其内部调用的也是execute()方法，因此我们重点分析execute()方法。源码奉上：
1 public void execute(Runnable command) { 2 if (command == null) 3 throw new NullPointerException(); 4 // 如果线程个数小于核心池大小，则创建核心线程 5 if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) { 6 // 如果线程个数大于等于核心池大小，则添加到任务队列 7 if (runState == RUNNING && workQueue.offer(command)) { 8 // 防止其他线程关闭了线程池，确保新加入的任务能得到执行 9 if (runState != RUNNING || poolSize == 0) 10 ensureQueuedTaskHandled(command); 11 //如果添加到任务队列失败（队列已满），且线程个数小于最大池，则创建最大池线程 12 } else if (!addIfUnderMaximumPoolSize(command)) 13 //如果任务队列和线程个数都达到上限，则采用拒绝策略 14 reject(command); // is shutdown or saturated 15 } 16 }
注意execute()方法调用的方法：
addInUderCorePoolSize
1 private boolean addIfUnderCorePoolSize(Runnable firstTask) { 2 Thread t = null; 3 final ReentrantLock mainLock = this.mainLock; 4 mainLock.lock(); 5 try { 6 // 同步判断线程个数是否小于核心池，小于则创建新线程 7 if (poolSize < corePoolSize && runState == RUNNING) 8 t = addThread(firstTask); 9 } finally { 10 mainLock.unlock(); 11 } 12 if (t == null) 13 return false; 14 t.start(); 15 return true; 16 }
addIfUnderMaximumPoolSize
private boolean addIfUnderMaximumPoolSize(Runnable firstTask) { Thread t = null; final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 同步判断线程个数是否小于最大池，小于则创建新线程 if (poolSize < maximumPoolSize && runState == RUNNING) t = addThread(firstTask); } finally { mainLock.unlock(); } if (t == null) return false; t.start(); return true; }
不难看出，addInUderCorePoolSize和addIfUnderMaximumPoolSize方法的实现代码非常相似，事实上他们的功能也基本相同———判断是否要创建新线程，如果需要则创建，否则返回false，区别在于创建线程的条件不同。并且，他们都调用了addThread()方法来创建线程，那么addThread()干了些什么呢？
private Thread addThread(Runnable firstTask) { // 创建一个工作任务Worker，实现了runnable接口 Worker w = new Worker(firstTask); // 使用worker创建工作线程 Thread t = threadFactory.newThread(w); if (t != null) { w.thread = t; // 将工作任务加入工作集 workers.add(w); // 线程个数增加 int nt = ++poolSize; if (nt > largestPoolSize) largestPoolSize = nt; } return t; }
上面的代码中，最重要是Worker对象，它实现了Runnable接口 ，并作为参数创建线程，看下Worker的主要代码：
1 private void runTask(Runnable task) { 2 final ReentrantLock runLock = this.runLock; 3 runLock.lock(); 4 try { 5 if (runState < STOP && 6 Thread.interrupted() && 7 runState >= STOP) 8 thread.interrupt(); 9 boolean ran = false; 10 // 可以通过重写此方法，做一些线程前置操作 11 beforeExecute(thread, task); 12 try { 13 task.run(); 14 ran = true; 15 // 可以通过重写此方法，做些线程后置操作 16 afterExecute(task, null); 17 ++completedTasks; 18 } catch (RuntimeException ex) { 19 if (!ran) 20 afterExecute(task, ex); 21 throw ex; 22 } 23 } finally { 24 runLock.unlock(); 25 } 26 } 27 28 public void run() { 29 try { 30 // 第一个task，也就是触发现成创建的task 31 Runnable task = firstTask; 32 firstTask = null; 33 // 循环从任务队列中取task 34 while (task != null || (task = getTask()) != null) { 35 // 执行task 36 runTask(task); 37 task = null; 38 } 39 } finally { 40 workerDone(this); 41 } 42 } 43 } 44 45 Runnable getTask() { 46 for (;;) { 47 try { 48 int state = runState; 49 // 如果是stop或者teminate状态，任务队列中的任务不会再处理，返回null 50 if (state > SHUTDOWN) 51 return null; 52 Runnable r; 53 // 如果是shutdown状态，处理完任务队列的任务，不会接受新任务 54 if (state == SHUTDOWN) 55 r = workQueue.poll(); 56 // 如果线程个数大于核心池或者核心线程允许超时，则取任务队列时加上超时时间 57 else if (poolSize > corePoolSize || allowCoreThreadTimeOut) 58 r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS); 59 // 否则取任务队列直至取到任务（阻塞） 60 else 61 r = workQueue.take(); 62 if (r != null) 63 return r; 64 if (workerCanExit()) { 65 if (runState >= SHUTDOWN) 66 interruptIdleWorkers(); 67 return null; 68 } 69 } catch (InterruptedException ie) { 70 } 71 } 72 }
上面的代码已经很清楚的展示线程池中每个工作线程是怎么运行的。工作线程启动后，首先执行本次提交的任务（代码中的firsttask），然后循环去任务队列读取新的待执行任务，可以根据线程池当前状态设置不同的读取队列元素的方式，如果读取队列失败，说明该工作线程处于空闲状态，可以关闭。
常见方法——计算密集型，设为CPU个数+1；IO密集型，设为2*CPU个数+1
精确计算——（ IO等待时间/CPU计算时间 + 1 ）* CPU个数
1 public class ThreadPool { 2 private LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(100000);; 3 private ThreadPoolExecutor threadPoolExecutor; 4 // 被拒绝的任务计数，可能有多个线程向同一个线程池中添加任务，保证线程安全 5 private AtomicInteger rejectTaskNum = new AtomicInteger(0); 6 private static ThreadPool instance; 7 8 private ThreadPool() { 9 // 拒绝策略，是ThreadPoolExecutor的静态内部类，AbortPolicy策略拒绝任务并抛出RejectedExecutionException异常 10 threadPoolExecutor = new ThreadPoolExecutor(2 * Runtime.getRuntime().availableProcessors(), 4 * Runtime.getRuntime().availableProcessors(), 11 1000L, TimeUnit.MILLISECONDS, queue, new ThreadPoolExecutor.AbortPolicy()); 12 } 13 14 public static ThreadPool getInstace() { 15 // 第一个判断为了提高效率，不用每次都走同步判断，大多数情况下instance不为null，直接返回instance 16 if (instance == null) { 17 // 多线程创建单例时保证线程安全 18 synchronized (ThreadPool.class) { 19 if (instance == null) { 20 instance = new ThreadPool(); 21 } 22 } 23 } 24 return instance; 25 } 26 27 /** 28 * 如果在添加任务的过程中抛出RejectedExecutionException异常 ，说明触发拒绝策略 29 * 30 * @param task 31 * @return 32 */ 33 public boolean addTask(Runnable task) { 34 boolean reject = false; 35 try { 36 threadPoolExecutor.execute(task); 37 } catch (RejectedExecutionException e) { 38 rejectTaskNum.getAndIncrement(); 39 reject = true; 40 } 41 return reject; 42 } 43 44 /** 45 * 获取线程池中队列的剩余容量大小 46 * 47 * @return 48 */ 49 public int getRemaining() { 50 return queue.remainingCapacity(); 51 } 52 53 /** 54 * 获取线程池中当前的线程个数 55 * 56 * @return 57 */ 58 public int getPoolSize() { 59 return threadPoolExecutor.getPoolSize(); 60 } 61 62 /** 63 * 获取线程池从启动开始拒绝的任务个数 64 * 65 * @return 66 */ 67 public int getRejectTaskNum() { 68 return rejectTaskNum.get(); 69 } 70 71 }
标签：
原文地址：http://www.cnblogs.com/summerautumn/p/5251811.html
