在 阿里巴巴Java开发手册 中有这么两段话,如下图所示
可以看到提到的两点,第一要求不能显示的创建线程,也就是new Thread的这种形式,需要使用线程池对线程进行管理,第二不允许使用官方提供的四种线程池,而是需要通过自行创建的方式去创建线程池,更加理解线程池的允许规则
先从构造函数看起:
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) 复制代码
有的朋友可能还不是很清晰,举个例子, 一个公司,核心线程就是代表公司的内部核心员工,最大线程数量就是员工的最大数量,可能包含非内部员工,因为有一些试点或者简单的项目,需要一些外协人员来做,也就是非核心线程,那么当这些项目做完了或者失败了,公司为了节约用人成本,就遣散非核心员工,也就是闲置线程的存活时间。假如核心员工每个人都很忙,但是需求又一波接一波,那就任务排期,也就是任务队列,当任务队列都满了时候,还要来需求?对不起,不接受,直接拒绝,这也就是handler对应的拒绝策略了 ,可以例子不是很合适,但是主要帮助大家理解下大概的意思。
打开源码类,可以看到如下几个变量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits 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; private static final int TERMINATED = 3 << COUNT_BITS; // Packing and unpacking ctl private static int runStateOf(int c) { return c & ~CAPACITY; } private static int workerCountOf(int c) { return c & CAPACITY; } private static int ctlOf(int rs, int wc) { return rs | wc; } 复制代码
AtomicInteger是一个原子操作类,保证线程安全,采用低29位表示线程的最大数量,高3位表示5种线程池状态,维护两个参数,workCount和runState。workCount表示有效的线程数量,runState表示线程池的运行状态。
terminated()
方法已经执行完成 引用一张图片帮助大家理解5个状态
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(); //如果当前线程数量小于核心线程数量,执行addWorker创建新线程执行command任务 if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } //如果当前是运行状态,将任务放入阻塞队列,double-check线程池状态 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); //如果再次check,发现线程池状态不是运行状态了,移除刚才添加进来的任务,并且拒绝改任务 if (! isRunning(recheck) && remove(command)) reject(command); //处于运行状态,但是没有线程,创建线程 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //往线程池中创建新的线程失败,则reject任务 else if (!addWorker(command, false)) reject(command); } 复制代码
这里大概总结下execute方法的执行流程,其实大家看源码方法注释是一样很好的学习方法
final void reject(Runnable command) { handler.rejectedExecution(command, this); } 复制代码
拒绝任务很简单,reject方法会调用handler的rejectedExecution(command,this)方法,handler是RejectedExecutionHandler接口,默认实现是AbortPolicy,下面是AbortPolicy的实现:
public static class AbortPolicy implements RejectedExecutionHandler { public AbortPolicy() { } public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString()); } } 复制代码
可以看到默认策略是直接抛出异常的,这只是默认使用的策略,可以通过实现接口实现自己的逻辑。
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 这里return false的情况有以下几种 //1.当前状态是stop及以上 2.当前是SHUTDOWN状态,但是firstTask不为空 //3.当前是SHUTDOWN状态,但是队列中为空 //从第一节我们知道,SHUTDOWN状态是不执行进来的任务的,但是会继续执行队列中的任务 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); 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.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; } 复制代码
这里就主要流程分析下
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** 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; Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); } ...... } 复制代码
可以看到,Worker内部维护,一个线程变量以及任务变量,启动一个 Worker对象中包含的线程 thread, 就相当于要执行 runWorker()方法, 并将该 Worker对象作为该方法的参数.
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //task不为空,执行当前任务,任务执行完后将task置位空,getTask方法接着不断从队列中取任务 while (task != null || (task = getTask()) != null) { w.lock(); //再次check线程池状态,如果是stop状态,直接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(); } 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++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } } 复制代码
通过while循环不断的调用getTask方法,获取任务task并进行执行,如果任务都执行完,跳出循环,线程结束并减少当前线程数量。
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? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; 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; } } } 复制代码
这里主要有两个判断需要说明下:
线程池中的细节比较多,大致做一下总结归纳
大概分析就是这么多,希望有能够帮助到一些朋友更好的理解线程池的工作原理以及在使用中能够更好的使用, 如有疑问或者错误,欢迎一起讨论