多线程（十七、深入了解线程池-ThreadPoolExecutor）
/** * 使用给定的参数创建ThreadPoolExecutor. * * @param corePoolSize 核心线程池中的最大线程数 * @param maximumPoolSize 总线程池中的最大线程数 * @param keepAliveTime 空闲线程的存活时间 * @param unit keepAliveTime的单位 * @param workQueue 任务队列, 保存已经提交但尚未被执行的任务 * @param threadFactory 线程工厂(用于指定如果创建一个线程) * @param handler 拒绝策略 (当任务太多导致工作队列满时的处理策略) */public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); // 使用纳秒保存存活时间 this.threadFactory = threadFactory; this.handler = handler;}
1、ThreadPoolExecutor在逻辑上将自身管理的线程池划分为两部分：核心线程池（大小对应为corePoolSize）、非核心线程池（大小对应为maximumPoolSize-corePoolSize）。
2、ThreadPoolExecutor中只有一种类型的线程，名叫Worker，它是ThreadPoolExecutor内部类，封装着Runnable任务和执行该任务的Thread对象，我们称它为【工作线程】，它也是ThreadPoolExecutor唯一需要进行维护的线程；
3、【核心线程池】【非核心线程池】都是逻辑上的概念，ThreadPoolExecutor在任务调度过程中会根据corePoolSize和maximumPoolSize的大小，判断如何执行任务。
1、低29位保存线程数
2、高3位保存线程池状态
//保存线程池状态和工作线程数:低29位: 工作线程数,高3位 : 线程池状态 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; // 最大线程数: 2^29-1 private static final int CAPACITY = (1 << COUNT_BITS) - 1; // 线程池状态 private static final int RUNNING = -1 << COUNT_BITS; //接受新任务, 且处理已经进入阻塞队列的任务 private static final int SHUTDOWN = 0 << COUNT_BITS; //不接受新任务, 但处理已经进入阻塞队列的任务 private static final int STOP = 1 << COUNT_BITS; //不接受新任务, 且不处理已经进入阻塞队列的任务, 同时中断正在运行的任务 private static final int TIDYING = 2 << COUNT_BITS; //所有任务都已终止, 工作线程数为0, 线程转化为TIDYING状态并准备调用terminated方法 private static final int TERMINATED = 3 << COUNT_BITS; //terminated方法已经执行完成工作线程工作线程（Worker），Worker内部类，实现了AQS框架，ThreadPoolExecutor通过一个HashSet来保存工作线程：
/** * Worker表示线程池中的一个工作线程, 可以与任务相关联. * 由于实现了AQS框架, 其同步状态值的定义如下: * -1: 初始状态 * 0: 无锁状态 * 1: 加锁状态 */private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * 与该Worker关联的线程. */ final Thread thread; /** *初始化任务，可以为空，为空的时候则去任务队列workQueue里获取 */ Runnable firstTask; /** * 当前工作线程处理完成的任务数 */ volatile long completedTasks; Worker(Runnable firstTask) { setState(-1); // 初始的同步状态值 this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** * 执行任务 */ public void run() { runWorker(this); } /** * 是否加锁 */ 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) { } } }}线程池执行executeexecute代码逻辑
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); //获取线程池状态和工作线程数量 int c = ctl.get(); //如果工作线程数 < 核心线程数，则创建工作线程 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) //addWorker创建工作线程 return; c = ctl.get(); } //工作线程创建失败，或者，工作线程 >= 核心线程数，任务插入任务队列workQueue 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); } //任务插入队列失败，创建非核心工作线程，如果失败，则说明工作线程 > 总线程数量，则执行拒绝策略 else if (!addWorker(command, false)) reject(command); }执行流程图
说明：
1、如果工作线程数小于核心线程池上限（CorePoolSize），则直接新建一个工作线程并执行任务；
2、如果工作线程数大于等于CorePoolSize，则尝试将任务加入到队列等待以后执行。如果队列已满，则在总线程池未满的情况下（CorePoolSize ≤ 工作线程数 ＜ maximumPoolSize）新建一个工作线程立即执行任务，否则执行拒绝策略。
/** * 添加工作线程并执行任务 * * @param firstTask 如果指定了该参数, 表示将立即执行该firstTask任务; 否则从工作队列中获取任务并执行 * @param core 执行任务的工作线程归属于哪个线程池: true-核心线程池 false-非核心线程池 */ private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); /** * 这个if主要是判断哪些情况下, 线程池不再接受新任务执行, 而是直接返回.总结下, 有以下几种情况： * 1. 线程池状态为 STOP 或 TIDYING 或 TERMINATED: 线程池状态为上述任一一种时, 都不会再接受任务，所以直接返回 * 2. 线程池状态≥ SHUTDOWN 且 firstTask != null: 因为当线程池状态≥ SHUTDOWN时, 不再接受新任务的提交，所以直接返回 * 3. 线程池状态≥ SHUTDOWN 且 队列为空: 队列中已经没有任务了, 所以也就不需要执行任何任务了，可以直接返回 */ if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { //获取工作线程数 int wc = workerCountOf(c); /** * 这个if主要是判断工作线程数是否超限, 以下任一情况属于超限, 直接返回: * 1. 工作线程数超过最大工作线程数(2^29-1) * 2. 工作线程数超过核心线程池上限(入参core为true, 表示归属核心线程池) * 3. 工作线程数超过总线程池上限(入参core为false, 表示归属非核心线程池) */ if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //增加工作线程数 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); //CAS失败，自旋重新操作 if (runStateOf(c) != rs) continue retry; } } 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 { //重新检查线程池状态 int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); //工作线程加入集合 workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { //启动线程，这个很关键，因为t在new的时候是用Work作为任务的， // Work实现了Runnale接口，所以t.start就是执行Work的run方法 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
特别注意：
启动线程，这个很关键，因为thread在new的时候是用Work（this）作为任务的, Work实现了Runnale接口，所以t.start就是执行Work的run方法。
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); // 任务, 如果是null则从队列取任务 Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // 允许执行线程被中断 boolean completedAbruptly = true; try { // 当task==null时会通过getTask从队列取任务 while (task != null || (task = getTask()) != null) { w.lock(); 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(); } 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; w.completedTasks++; // 完成任务数+1 w.unlock(); } } //说明该工作线程自身既没有携带任务, 也没从任务队列中获取到任务 completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }
获取任务方法getTask
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); /** * 以下IF用于判断哪些情况下不允许再从队列获取任务: * 1. 线程池进入停止状态（STOP/TIDYING/TERMINATED）, 此时即使队列中还有任务未执行, 也不再执行 * 2. 线程池非RUNNING状态, 且队列为空 */ if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); /** * timed变量用于判断是否需要进行超时控制: * 对于核心线程池中的工作线程, 除非设置了allowCoreThreadTimeOut==true, 否则不会超时回收; * 对于非核心线程池中的工作线程, 都需要超时控制 */ boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 这里主要是当外部通过setMaximumPoolSize方法重新设置了最大线程数时 // 需要回收多出的工作线程 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { //区分超时操作还是非超时获取 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }拒绝策略
所谓拒绝策略，就是在构造ThreadPoolExecutor时，传入的RejectedExecutionHandler对象，一共4种
1.AbortPolicy（默认），抛出一个RejectedExecutionException异常
2、DiscardPolicy，什么都不做，等任务自己被回收
3、DiscardOldestPolicy，丢弃任务队列中的最近一个任务，并执行当前任务
4、CallerRunsPolicy，以自身线程来执行任务,减缓新任务提交的速度
声明：本站所有文章资源内容，如无特殊说明或标注，均为采集网络资源。如若本站内容侵犯了原著者的合法权益，可联系本站删除。
