目录
线程池
一、Java构建线程的方法
二、线程池的7个参数
三、线程池的执行流程
四、线程池属性标识
4.1.核心属性
4.2.线程池状态变化
五、线程池的 execute 方法执行
六、Worker的封装
七、线程执行的后续处理
线程池
线程池ThreadPoolExecutor源码剖析
// ExecutorService executorService = Executors.newFixedThreadPool(1);ThreadPoolExecutor executor = new ThreadPoolExecutor(2,4,10,TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(2),new ThreadFactory() {@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r);t.setName("abc");return t;}},new ThreadPoolExecutor.AbortPolicy());//线程池执行任务executor.execute(() ->{for (int i = 0; i < 10; i++) {System.out.println(i);}});
一、Java构建线程的方法
- 实现Runnable
- 继承Thread
- 实现Callable
- 有返回值return，可以抛出异常
- 线程池方式
- 优点：避免频繁创建线程和销毁线程带来的损耗
- Java提供了构建线程池的的方式，用 Executors 可以去创建（但是规范中不允许使用这种方式创建线程池，这种方式对线程的控制力度比较低）
- 推荐手动创建线程池
二、线程池的7个参数
七个参数：
| 参数 | 解释 |
| corePoolSize | 核心线程数 |
| maximumPoolSize | 最大线程数 |
| keepAliveTime | 最大空闲时间 |
| TimeUnit | 时间单位 |
| BlockingQueue<Runnable> | 阻塞队列 |
| ThreadFactory | 线程工厂 |
| RejectedExecutionHandler | 拒绝策略 |
三、线程池的执行流程
为什么要先进阻塞再去尝试创建非核心线程：
eg：饭店（线程池）--- 厨师（线程）---人多先排队（阻塞队列）---招厨师（创建最大线程数）---客满（拒绝策略）
四、线程池属性标识
4.1.核心属性
//是一个int类型的数值，表达了两个意思，1：声明当前线程池的状态，2：声明线程池中的线程数
//高3位：线程池状态 低29位：线程池中的线程个数
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); //固定值，29，方便后面做位运算 private static final int COUNT_BITS = Integer.SIZE - 3; //通过位运算得出最大容量 private static final int CAPACITY = (1 << COUNT_BITS) - 1;// 线程池状态 //111 代表线程池为 RUNNING ，代表正常接收任务 private static final int RUNNING = -1 << COUNT_BITS; //000 代表线程池为 SHUTDOWN 状态，不接收新任务，但是内部还会处理阻塞队列中的任务，正在进行的任务也正常处理 private static final int SHUTDOWN = 0 << COUNT_BITS; //001 代表线程池为 STOP 状态，不接收新任务，也不去处理阻塞队列中的任务，同时会中断正在执行的任务 private static final int STOP = 1 << COUNT_BITS; //010 代表线程池为 TIDYING 状态，过渡的状态，代表当前线程池即将Game over private static final int TIDYING = 2 << COUNT_BITS; //011 代表线程池为 TERMINATED ，凉透了 private static final int TERMINATED = 3 << COUNT_BITS;// 得到线程池的状态 private static int runStateOf(int c) { return c & ~CAPACITY; } // 得到当前线程池的线程数量 private static int workerCountOf(int c) { return c & CAPACITY; }
4.2.线程池状态变化
线程池的状态可以通过ThreadPoolExecutor类的getPoolSize()方法，getActiveCount()方法，getCompletedTaskCount()方法和getTaskCount()方法来查询。线程池的状态有以下几种：
1. Running（运行状态）：线程池新建或调用execute()方法后，处于运行状态，能够接收新的任务。
2. Shutdown（关闭状态）：线程池调用shutdown()方法后，线程池的状态会变为Shutdown。此时线程池不再接收新的任务，但会执行已提交的等待任务队列中的任务。
3. Stop（停止状态）：人为调用shutdownNow()方法后，线程池的状态会变为Stop。此时线程池不再接收新的任务，并且会中断正在处理中的任务。
4. Tidying（整理状态）：当线程池处于Shutdown或Stop状态时，如果等待队列中还有未执行的任务，则线程池将执行整理操作，将等待队列中的未执行任务移除，并保存到一个列表中。
5. Terminated（终止状态）：当线程池处于Shutdown状态，并且等待队列中的任务全部执行完毕，或者在Stop状态下，线程池内部的所有线程都已经终止时，线程池进入Terminated状态。
线程池的状态变化如上所述，可以更好地对线程池进行管理和监控。
五、线程池的 execute 方法执行
从execute方法开始
public void execute(Runnable command) { //健壮性判断if (command == null)throw new NullPointerException();//拿到32位的intint c = ctl.get(); //获取 工作线程数 < 核心线程数if (workerCountOf(c) < corePoolSize) {//进到if，代表可以创建核心线程数if (addWorker(command, true)) //到这结束return; //如果if没进去，代表创建核心线程数失败，重新获取ctl 32位的intc = ctl.get();} //判断线程池是不是RUNNING,将任务添加到阻塞队列中的if (isRunning(c) && workQueue.offer(command)) { //再次获取ctlint recheck = ctl.get(); //再次判断是否是RUNNING， 如果不是RUNNING，移除任务if (! isRunning(recheck) && remove(command))reject(command);//拒绝策略//如果线程处在 RUNNING 状态，但是工作线程为0else if (workerCountOf(recheck) == 0)addWorker(null, false);//阻塞队列有任务，但是没有工作线程，添加一个任务为空的工作线程处理阻塞队列中的任务}//创建非核心线程，处理任务else if (!addWorker(command, false))reject(command);//拒绝策略
}
通过上述源码，掌握了线程池的执行流程，再次查看 addWorker 方法内部做了什么处理。
private boolean addWorker(Runnable firstTask, boolean core) { //标记for循环 retry: //经过大量的判断，给工作线程数标识 +1；for (;;) { //获取ctl （32位）int c = ctl.get(); //获取线程池状态int rs = runStateOf(c);if (rs >= SHUTDOWN && // 除了RUNNING都有可能! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()) // rs == SHUTDOWN ，就代表是STOP或者更高的状态，这时，不需要添加线程处理任务// 任务为空，如果任务位 null，并且线程池状态不是 RUNNING ,不需要处理// 阻塞队列不为null ，如果阻塞队列为空，返回 false ，外侧的 ！ 再次取反，获取 true，不需要处理 ) //构建工作线程失败！ return false;for (;;) { //获取工作线程个数int wc = workerCountOf(c);
//如果当前线程已经大于线程池最大容量，不去创建了 if (wc >= CAPACITY || // 判断wc是否超过核心线程或者最大线程 wc >= (core ? corePoolSize : maximumPoolSize))//构建工作线程失败！ return false; // 将工作线程数+1 ， 采用CAS的方式 if (compareAndIncrementWorkerCount(c)) //成功就退出外侧 for 循环break retry;// 重新获取ctl c = ctl.get();
//重新判断线程池状态 如果状态没变化：重新执行内部循环即可if (runStateOf(c) != rs) //如果状态有变化：结束这次外侧循环，开始下次外侧循环 continue retry;} } // worker 开始boolean workerStarted = false;boolean workerAdded = false; //Worker 就是工作线程Worker w = null;try { //创建Worker，传入任务w = new Worker(firstTask); //从Worker中获取线程tfinal Thread t = w.thread;
//如果线程 t 不为nullif (t != null) { //获取线程池的全局锁，避免我添加任务时，其他线程干掉了线程池，干掉线程池需要先获得这个锁final ReentrantLock mainLock = this.mainLock; //加锁mainLock.lock();try { //获取线程池状态int rs = runStateOf(ctl.get());//是RUNNING状态if (rs < SHUTDOWN || //是SHUTDOWN状态，创建空任务工作线程，处理阻塞队列中的任务(rs == SHUTDOWN && firstTask == null)) { //线程是否是运行状态
if (t.isAlive()) throw new IllegalThreadStateException();//将工作线程添加到集合中workers.add(w);//获取工作线程个数int s = workers.size();//如果现在工作线程数，大于之前记录的最大工作线程数，就替换一下if (s > largestPoolSize)largestPoolSize = s;//workerAdded 为true，添加工作线程成功workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {//启动工作线程t.start(); //启动工作线程成功workerStarted = true;}}} finally { //如果启动工作线程失败，就调用下面的方法if (! workerStarted)addWorkerFailed(w);}//返回工作是否启动return workerStarted;
}
六、Worker的封装
Worker的封装（源码）
private final class Workerextends AbstractQueuedSynchronizerimplements Runnable
{private static final long serialVersionUID = 6138294804551838833L;final Thread thread;Runnable firstTask;volatile long completedTasks;Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this);}public void run() {runWorker(this);}
看 runWorker 方法
final void runWorker(Worker w) {
//获取当前线程 Thread wt = Thread.currentThread();//拿到任务Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interrupts//标识为trueboolean completedAbruptly = true;try {//任务不空，执行任务。 如果任务为空，通过getTask从阻塞队列中获取任务！while (task != null || (task = getTask()) != null) {//加锁，避免你shutdown我任务也不会中断w.lock();//获取当前状态，是否大于等于STOP,被拒！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);}
}
