详解Java线程池是如何重复利用空闲线程的

 public static ExecutorService newFixedThreadPool(int var0) { return new ThreadPoolExecutor(var0, var0, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()); } public static ExecutorService newSingleThreadExecutor() { return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue())); } public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, 2147483647, 60L, TimeUnit.SECONDS, new SynchronousQueue()); } public static ScheduledExecutorService newSingleThreadScheduledExecutor() { return new Executors.DelegatedScheduledExecutorService(new ScheduledThreadPoolExecutor(1)); } public static ScheduledExecutorService newScheduledThreadPool(int var0) { return new ScheduledThreadPoolExecutor(var0); }

 /** * Creates a new {@code ThreadPoolExecutor} with the given initial * parameters. * * @param corePoolSize the number of threads to keep in the pool, even *        if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the *        pool * @param keepAliveTime when the number of threads is greater than *        the core, this is the maximum time that excess idle threads *        will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are *        executed.  This queue will hold only the {@code Runnable} *        tasks submitted by the {@code execute} method. * @param threadFactory the factory to use when the executor *        creates a new thread * @param handler the handler to use when execution is blocked *        because the thread bounds and queue capacities are reached * @throws IllegalArgumentException if one of the following holds:
 *         {@code corePoolSize <0}
 *         {@code keepAliveTime <0}
 *         {@code maximumPoolSize <= 0}
 *         {@code maximumPoolSize  workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize <0 || maximumpoolsize < corepoolsize keepalivetime 0) throw new illegalargumentexception(); if (workqueue== null threadfactory== handler== null) nullpointerexception(); this.corepoolsize=corePoolSize; this.maximumpoolsize=maximumPoolSize; this.workqueue=workQueue; this.keepalivetime=unit.toNanos(keepAliveTime); this.threadfactory=threadFactory; this.handler=handler; }< pre>

 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) 

 // 获取运行状态 private static int runStateOf(int c)     { return c & ~CAPACITY; } // 获取活动线程数 private static int workerCountOf(int c)  { return c & CAPACITY; }

 private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c); if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get();  // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } //前面都是线程池状态的判断，暂时不理会，主要看下面两个关键的地方 boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); // 新建一个Worker对象，这个对象包含了待执行的任务，并且新建一个线程 final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get()); if (rs  largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { t.start(); // 启动刚创建的worker对象里面的thread执行 workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }

 private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ private static final long serialVersionUID = 6138294804551838833L; /** Thread this worker is running in.  Null if factory fails. */ final Thread thread; /** Initial task to run.  Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */ Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker. */ public void run() { runWorker(this); } // Lock methods // // The value 0 represents the unlocked state. // The value 1 represents the locked state. protected boolean isHeldExclusively() { return getState() != 0; } protected boolean tryAcquire(int unused) { if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } protected boolean tryRelease(int unused) { setExclusiveOwnerThread(null); setState(0); return true; } public void lock()        { acquire(1); } public boolean tryLock()  { return tryAcquire(1); } public void unlock()      { release(1); } public boolean isLocked() { return isHeldExclusively(); } void interruptIfStarted() { Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { t.interrupt(); } catch (SecurityException ignore) { } } } }

 Worker(Runnable firstTask) { // worker本身实现了Runnable接口 setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; // 持有外部传进来的runnable任务 //创建了一个thread对象，并把自身这个runnable对象给了thread，一旦该thread执行start方法，就会执行worker的run方法 this.thread = getThreadFactory().newThread(this); } 在addWorker方法中执行的t.start会去执行worker的run方法： public void run() { runWorker(this); } run方法又执行了ThreadPoolExecutor的runWorker方法，把当前worker对象传入。 final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; // 取出worker的runnable任务 w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { // 循环不断的判断任务是否为空，当第一个判断为false的时候，即task为null，这个task啥时候为null呢？ // 要么w.firstTask为null，还记得我们在execute方法第二步的时候，执行addWorker的时候传进来的runnable是null吗？ // 要么是执行了一遍while循环，在下面的finally中执行了task=null； // 或者执行第二个判断，一旦不为空就会继续执行循环里的代码。 while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted.  This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { task.run(); // 任务不为空，就会执行任务的run方法，也就是runnable的run方法 } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { afterExecute(task, thrown); } } finally { task = null; // 执行完成置null，继续下一个循环 w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }

 private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // timed变量用于判断是否需要进行超时控制。 // allowCoreThreadTimeOut默认是false，也就是核心线程不允许进行超时； // wc > corePoolSize，表示当前线程池中的线程数量大于核心线程数量； // 对于超过核心线程数量的这些线程或者允许核心线程进行超时控制的时候，需要进行超时控制 // Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 如果需要进行超时控制，且上次从缓存队列中获取任务时发生了超时（timedOut开始为false，后面的循环末尾超时时会置为true) // 或者当前线程数量已经超过了最大线程数量，那么尝试将workerCount减1,即当前活动线程数减1， if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { // 如果减1成功，则返回null，这就意味着runWorker()方法中的while循环会被退出，其对应的线程就要销毁了，也就是线程池中少了一个线程了 if (compareAndDecrementWorkerCount(c)) return null; continue; } try { // 注意workQueue中的poll()方法与take()方法的区别 //poll方式取任务的特点是从缓存队列中取任务,最长等待keepAliveTime的时长，取不到返回null //take方式取任务的特点是从缓存队列中取任务，若队列为空,则进入阻塞状态，直到能取出对象为止 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; // 能走到这里说明已经超时了 } catch (InterruptedException retry) { timedOut = false; } } }