上一篇：面试题：fixed线程池原理及采用方法你了解吗？
一、demo
public class ThreadPoolExecutorTest { public static void main(String[] args) throws Exception{ ExecutorService executorService = Executors.newCachedThreadPool(); for (int i = 0; i < 10; i++){ executorService.execute(new Runnable() { @Override public void run() { try { System.out.println("线程池异步执行任务，线程=" + Thread.currentThread()); Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } }); } } }
二、构造方法
corePoolSize(核心线程数)：0
maximumPoolSize(非核心线程数或者最大线程数)：Integer.MAX_VALUE
keepAliveTime（非核心线程空闲多久销毁）：60s
workQueue（任务队列）：SynchronousQueue
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }
public SynchronousQueue() { this(false); }
private transient volatile Transferer<E> transferer; public SynchronousQueue(boolean fair) { /** Transferer的实现有公平和非公平的，默认是非公平的：TransferStack * TransferQueue：队列数据结构，先入先出 * TransferStack：栈数据结构，先入后出 */ transferer = fair ? new TransferQueue<E>() : new TransferStack<E>(); }
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
三、execute方法
在向线程池提交任务时，因为核心线程数为0所以会去队列里面添加元素，第一次队列头节点为空，添加失败则在会去创建非核心线程去执行任务
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(); /** 判断线程池中的线程数量是否小于corePoolSize，小于则创建线程进行执行 */ if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } /** 判断线程池状态是否运行，如果运行则将任务加入队列，加入队列失败则返回false */ if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); /** 判断线程池是否运行，不运行则从队列中删除任务 */ if (! isRunning(recheck) && remove(command)) /** 拒绝任务 */ reject(command); /** 判断线程次线程数量，如果为0则说明队列数据无法消费，创建线程进行处理 */ else if (workerCountOf(recheck) == 0) addWorker(null, false); } /** 添加非核心线程，如果已达到最大非核心线程数则是否，调用reject方法进行拒绝 */ else if (!addWorker(command, false)) reject(command); }
根据ThreadPoolExecutor构造时传入的队列实现为TransferStack非公平队列，我们就看一下这个队列添加元素时是怎么实现的，刚刚开始添加元素肯定是为false，所以会创建一个非核心线程去处理任务；
public boolean offer(E e) { if (e == null) throw new NullPointerException(); return transferer.transfer(e, true, 0) != null; }
static final int DATA = 1; E transfer(E e, boolean timed, long nanos) { /* * Basic algorithm is to loop trying one of three actions: * * 1. If apparently empty or already containing nodes of same * mode, try to push node on stack and wait for a match, * returning it, or null if cancelled. * * 2. If apparently containing node of complementary mode, * try to push a fulfilling node on to stack, match * with corresponding waiting node, pop both from * stack, and return matched item. The matching or * unlinking might not actually be necessary because of * other threads performing action 3: * * 3. If top of stack already holds another fulfilling node, * help it out by doing its match and/or pop * operations, and then continue. The code for helping * is essentially the same as for fulfilling, except * that it doesn't return the item. */ SNode s = null; // constructed/reused as needed /** 这里的e为我们传递进来的任务，所以这里则为DATA，DATA：1*/ int mode = (e == null) ? REQUEST : DATA; for (;;) { /** h指向head头节点，SNode为单向链表结构 */ SNode h = head; /** head刚刚开始为空，所以会进行逻辑处理 * 如果有线程等待的话则head不为空，会走else分支 */ if (h == null || h.mode == mode) { // empty or same-mode /** timed为true，nanos为0，满足条件进行处理 * 后续如果是非核心线程去获取任务时会调用poll方法进行元素获取 * poll方法会存入一个超时时间，nanos则会大于0走else分支 */ if (timed && nanos <= 0) { // can't wait /** 第一次head头节点为空，直接返回null */ if (h != null && h.isCancelled()) casHead(h, h.next); // pop cancelled node else return null; /** 设置新创建的snode节点为头节点 */ } else if (casHead(h, s = snode(s, e, h, mode))) { /** 阻塞等待，直到有线程进行put数据进行匹配完成之后进行唤醒 * nanos不为空，默认为60秒，会阻塞等待60秒 * 如果60秒没有线程来匹配的话则会销毁Worker */ SNode m = awaitFulfill(s, timed, nanos); if (m == s) { // wait was cancelled clean(s); return null; } if ((h = head) != null && h.next == s) casHead(h, s.next); // help s's fulfiller return (E) ((mode == REQUEST) ? m.item : s.item); } /** 因为之前是poll方法进行调用，所以mode为0，因此会进入分支处理 */ } else if (!isFulfilling(h.mode)) { // try to fulfill if (h.isCancelled()) // already cancelled casHead(h, h.next); // pop and retry else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) { for (;;) { // loop until matched or waiters disappear SNode m = s.next; // m is s's match if (m == null) { // all waiters are gone casHead(s, null); // pop fulfill node s = null; // use new node next time break; // restart main loop } SNode mn = m.next; /** 尝试匹配 */ if (m.tryMatch(s)) { /** 匹配成功之后将下一个节点设置为Head头节点 */ casHead(s, mn); // pop both s and m /** 返回元素 */ return (E) ((mode == REQUEST) ? m.item : s.item); } else // lost match s.casNext(m, mn); // help unlink } } } else { // help a fulfiller SNode m = h.next; // m is h's match if (m == null) // waiter is gone casHead(h, null); // pop fulfilling node else { SNode mn = m.next; if (m.tryMatch(h)) // help match casHead(h, mn); // pop both h and m else // lost match h.casNext(m, mn); // help unlink } } } }
/** *旋转/阻塞，直到节点s与完成操作匹配。 *@param s等待节点 *@param timed true如果已计时等待 *@param nanos超时值 *@返回匹配的节点，如果取消则返回s */ SNode awaitFulfill(SNode s, boolean timed, long nanos) { /* * When a node/thread is about to block, it sets its waiter * field and then rechecks state at least one more time * before actually parking, thus covering race vs * fulfiller noticing that waiter is non-null so should be * woken. * * When invoked by nodes that appear at the point of call * to be at the head of the stack, calls to park are * preceded by spins to avoid blocking when producers and * consumers are arriving very close in time. This can * happen enough to bother only on multiprocessors. * * The order of checks for returning out of main loop * reflects fact that interrupts have precedence over * normal returns, which have precedence over * timeouts. (So, on timeout, one last check for match is * done before giving up.) Except that calls from untimed * SynchronousQueue.{poll/offer} don't check interrupts * and don't wait at all, so are trapped in transfer * method rather than calling awaitFulfill. */ final long deadline = timed ? System.nanoTime() + nanos : 0L; Thread w = Thread.currentThread(); int spins = (shouldSpin(s) ? (timed ? maxTimedSpins : maxUntimedSpins) : 0); for (;;) { if (w.isInterrupted()) s.tryCancel(); SNode m = s.match; if (m != null) return m; if (timed) { nanos = deadline - System.nanoTime(); if (nanos <= 0L) { s.tryCancel(); continue; } } if (spins > 0) spins = shouldSpin(s) ? (spins-1) : 0; else if (s.waiter == null) s.waiter = w; // establish waiter so can park next iter else if (!timed) LockSupport.park(this); else if (nanos > spinForTimeoutThreshold) LockSupport.parkNanos(this, nanos); } }
四、addWorker向线程池添加Worker执行线程
private final HashSet<Worker> workers = new HashSet<Worker>(); 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); /** 判断线程池线程数量是否达到corePoolSize数量或者maximumPoolSize数量 */ if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; /** 通过CAS安全的来递增当前线程的数量 * 失败则break进行下一轮循环再次尝试递增线程数 */ 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); 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 < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); /** 将Worker线程放入workers的HashSet结构中 */ workers.add(w); /** 获取workers的大小，如果大于largestPoolSize则进行更新 * largestPoolSize：线程池当前线程数量最大大小 */ int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; /** 线程成功放入线程池 */ workerAdded = true; } } finally { /** 释放锁 */ mainLock.unlock(); } if (workerAdded) { /** 启动线程 */ t.start(); /** 线程启动标志位设置为true */ workerStarted = true; } } } finally { /** 线程启动失败则从线程池中删除失败线程 */ if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
private boolean compareAndIncrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect + 1); }
Worker就是工作线程，他是一个AQS的子类，本身自己就可以直接基于AQS来实现独占锁的机制，在Worker的run方法里面负责执行你提交的Runnable任务，是一个组件，里面保存了一个核心的state，代表了工作线程的状态；
Worker内部是基于threadFactory（有一个默认的线程工厂），你也可以自己手动指定一个线程工厂，此时就会按照线程工厂的策略来创建一个线程，放在Worker的内部，代表了执行任务的工作线程；
Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; /** 这里会将Worker自己的Runnable传递进去，线程的run方法其实就是Worker的run方法 */ this.thread = getThreadFactory().newThread(this); }
上面Worker初始化方法里面会调用ThreadFactory去创建线程，工厂是构建ThreadPoolExecutor时传入进去的，默认是DefaultThreadFactory，所以我们看一下DefaultThreadFactory的newThread方法；
static class DefaultThreadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix; DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } public Thread newThread(Runnable r) { /** 构建线程，线程组和线程前缀在DefaultThreadFactory构建的时候生成 */ Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) /** 线程为非后台线程 */ t.setDaemon(false); /** 设置线程优先级 */ if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); return t; } }
private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (w != null) workers.remove(w); decrementWorkerCount(); /** 尝试关闭线程 */ tryTerminate(); } finally { mainLock.unlock(); } }
final void tryTerminate() { for (;;) { int c = ctl.get(); if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; if (workerCountOf(c) != 0) { // Eligible to terminate interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { terminated(); } finally { ctl.set(ctlOf(TERMINATED, 0)); termination.signalAll(); } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } }
三、Worker线程的run方法
public void run() { runWorker(this); }
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); /** 这里的Runnable就是我们传递进来的Runnable */ Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { /** 先判断task是否为空，如果不为空则加锁执行任务 * 如果为空则调用getTask从队列中获取任务执行 * 如果getTask也是空则跳出循环执行线程销毁逻辑 */ 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 { /** 执行之前做一些事情，线程池ThreadPoolExecutor的扩展点 */ beforeExecute(wt, task); Throwable thrown = null; try { /** 执行Runnable的run方法 */ task.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 { /** 线程执行之后做一些事情，线程池ThreadPoolExecutor的扩展点 */ afterExecute(task, thrown); } } finally { task = null; /** Worker执行了多少次 */ w.completedTasks++; /** 释放锁 */ w.unlock(); } } completedAbruptly = false; } finally { /** 这里会做一些线程异常的处理，并且会将Worker线程从线程池中删除 */ processWorkerExit(w, completedAbruptly); } }
private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { completedTaskCount += w.completedTasks; /** 移除Worker */ workers.remove(w); } finally { mainLock.unlock(); } tryTerminate(); /** 判断线程池状态 */ int c = ctl.get(); if (runStateLessThan(c, STOP)) { if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) min = 1; /** 判断线程池线程数量是否大于等于min，大于直接返回 * 相当于当前线程数量 < corePoolSize，就会重新创建一个Worker和线程放入线程池中 */ if (workerCountOf(c) >= min) return; // replacement not needed } addWorker(null, false); } }
getTask方法向任务等待队列中获取任务，因为cached线程池的corePoolSize为0，因此timed为true，所以采用的workQueue.poll去获取任务
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); // Are workers subject to culling? /** 是否允许线程超时 * 配置项（allowCoreThreadTimedOut）默认：false * 当前线程数量超过了corePoolSize */ boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; /** 判断线程是否可以超时，线程池的线程数量是否大于1，线程等待队列是否为空 */ if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { /** 尝试将线程池数量减一 */ if (compareAndDecrementWorkerCount(c)) return null; continue; } try { /** 根据timed是否可以超时去调用workQueue的不同方法 * poll方法：获取元素，达到一点时间则返回null * take方法：阻塞式的LinkedBlockingQueue里获取任务 */ Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) /** 返回任务 */ return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
public E poll(long timeout, TimeUnit unit) throws InterruptedException { E e = transferer.transfer(null, true, unit.toNanos(timeout)); if (e != null || !Thread.interrupted()) return e; throw new InterruptedException(); }
总结
1、系统高峰期导致大量的线程被创建出来，会导致机器的CPU负载过高，甚至线程太多导致内存溢出，线程太多了，导致CPU负载飙升。队列无限增长，内存飙升；
2、cached线程池基本上不会排队任务可以立即执行，如果有线程在等待获取任务执行则会直接执行，否则就会创建一个非核心线程去执行；
3、无论cached线程池还是fixed线程池都不会触发拒绝任务提交的机制，因为cached线程池队列添加任务失败了则会直接创建非核心线程，这个数量无限大，所以理论上可以一直添加任务；
还没有评论，来说两句吧...