线程池 一时失言乱红尘 2022-05-28 07:50 116阅读 0赞 # 线程池 # * 线程池 * 线程池概述 * 创建一个线程池并提交线程任务 * 线程池源码解析 * 参数认识 * 构造方法 * 提交任务 * addWorker * 执行任务 * 关闭线程池 * 例子 ## 线程池概述 ## * 什么是线程池 * 为什么使用线程池 * 线程池的优势 * 第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。 * 第二：提高响应速度。当任务到达时，任务可以不需要的等到线程创建就能立即执行。 * 第三：提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。但是要做到合理的利用线程池，必须对其原理了如指掌。 ## 创建一个线程池并提交线程任务 ## ## 线程池源码解析 ## ### 参数认识 ### 1. corePoolSize : 线程池的基本大小，当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads方法，线程池会提前创建并启动所有基本线程。 2. runnableTaskQueue：任务对列，用于保存等待执行的任务的阻塞队列。可以选择以下几个阻塞队列。 * ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序。 * LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。 * SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。 * PriorityBlockingQueue：一个具有优先级得无限阻塞队列。 3. maximumPoolSize：线程池最大大小，线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是如果使用了无界的任务队列这个参数就没什么效果。 4. ThreadFactory：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字，Debug和定位问题时非常又帮助。 5. RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。 * CallerRunsPolicy：只用调用者所在线程来运行任务。 * DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。 * DiscardPolicy：不处理，丢弃掉。 * 当然也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略。如记录日志或持久化不能处理的任务。 6. keepAliveTime ：线程活动保持时间，线程池的工作线程空闲后，保持存活的时间。所以如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率。 7. TimeUnit：线程活动保持时间的单位，可选的单位有天（DAYS），小时（HOURS），分钟（MINUTES），毫秒(MILLISECONDS)，微秒(MICROSECONDS, 千分之一毫秒)和毫微秒(NANOSECONDS, 千分之一微秒)。 8. 类中其他属性 // 线程池的控制状态:用来表示线程池的运行状态（整型的高3位）和运行的worker数量（低29位） private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 29位的偏移量 private static final int COUNT_BITS = Integer.SIZE - 3; // 最大容量（2^29 - 1） private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits // 线程运行状态，总共有5个状态，需要3位来表示（所以偏移量的29 = 32 - 3） /** * RUNNING : 接受新任务并且处理已经进入阻塞队列的任务 * SHUTDOWN ： 不接受新任务，但是处理已经进入阻塞队列的任务 * STOP : 不接受新任务，不处理已经进入阻塞队列的任务并且中断正在运行的任务 * TIDYING : 所有的任务都已经终止，workerCount为0， 线程转化为TIDYING状态并且调用terminated钩子函数 * TERMINATED: terminated钩子函数已经运行完成 **/ 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; // 阻塞队列 private final BlockingQueue<Runnable> workQueue; // 可重入锁 private final ReentrantLock mainLock = new ReentrantLock(); // 存放工作线程集合 private final HashSet<Worker> workers = new HashSet<Worker>(); // 终止条件 private final Condition termination = mainLock.newCondition(); // 最大线程池容量 private int largestPoolSize; // 已完成任务数量 private long completedTaskCount; // 线程工厂 private volatile ThreadFactory threadFactory; // 拒绝执行处理器 private volatile RejectedExecutionHandler handler; // 线程等待运行时间 private volatile long keepAliveTime; // 是否运行核心线程超时 private volatile boolean allowCoreThreadTimeOut; // 核心池的大小 private volatile int corePoolSize; // 最大线程池大小 private volatile int maximumPoolSize; // 默认拒绝执行处理器 private static final RejectedExecutionHandler defaultHandler = new AbortPolicy(); ### 构造方法 ### public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize < 0 || // 核心大小不能小于0 maximumPoolSize <= 0 || // 线程池的初始最大容量不能小于0 maximumPoolSize < corePoolSize || // 初始最大容量不能小于核心大小 keepAliveTime < 0) // 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. 如果运行的线程小于corePoolSize,则尝试使用用户定义的Runnalbe对象创建一个新的线程 * 调用addWorker函数会原子性的检查runState和workCount，通过返回false来防止在不应 * 该添加线程时添加了线程 * 2. 如果一个任务能够成功入队列，在添加一个线城时仍需要进行双重检查（因为在前一次检查后 * 该线程死亡了），或者当进入到此方法时，线程池已经shutdown了，所以需要再次检查状态， * 若有必要，当停止时还需要回滚入队列操作，或者当线程池没有线程时需要创建一个新线程 * 3. 如果无法入队列，那么需要增加一个新线程，如果此操作失败，那么就意味着线程池已经shut * down或者已经饱和了，所以拒绝任务 */ public void execute(Runnable command) { if (command == null) throw new NullPointerException(); // 获取线程池控制状态 int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { // worker数量小于corePoolSize if (addWorker(command, true)) // 添加worker // 成功则返回 return; // 不成功则再次获取线程池控制状态 c = ctl.get(); } // 线程池处于RUNNING状态，将用户自定义的Runnable对象添加进workQueue队列 if (isRunning(c) && workQueue.offer(command)) { // 再次检查，获取线程池控制状态 int recheck = ctl.get(); // 线程池不处于RUNNING状态，将自定义任务从workQueue队列中移除 if (! isRunning(recheck) && remove(command)) // 拒绝执行命令 reject(command); else if (workerCountOf(recheck) == 0) // worker数量等于0 // 添加worker addWorker(null, false); } else if (!addWorker(command, false)) // 添加worker失败 // 拒绝执行命令 reject(command); } #### addWorker #### 1. 原子性的增加workerCount。 2. 将用户给定的任务封装成为一个worker，并将此worker添加进workers集合中。 3. 启动worker对应的线程，并启动该线程，运行worker的run方法。 4. 回滚worker的创建动作，即将worker从workers集合中删除，并原子性的减少workerCount。 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 && // 状态大于等于SHUTDOWN，初始的ctl为RUNNING，小于SHUTDOWN ! (rs == SHUTDOWN && // 状态为SHUTDOWN firstTask == null && // 第一个任务为null ! workQueue.isEmpty())) // worker队列不为空 // 返回 return false; for (;;) { // worker数量 int wc = workerCountOf(c); if (wc >= CAPACITY || // worker数量大于等于最大容量 wc >= (core ? corePoolSize : maximumPoolSize)) // worker数量大于等于核心线程池大小或者最大线程池大小 return false; if (compareAndIncrementWorkerCount(c)) // 比较并增加worker的数量 // 跳出外层循环 break retry; // 获取线程池控制状态 c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) // 此次的状态与上次获取的状态不相同 // 跳过剩余部分，继续循环 continue retry; // else CAS failed due to workerCount change; retry inner loop } } // worker开始标识 boolean workerStarted = false; // worker被添加标识 boolean workerAdded = false; // Worker w = null; try { // 初始化worker w = new Worker(firstTask); // 获取worker对应的线程 final Thread t = w.thread; if (t != null) { // 线程不为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 || // 小于SHUTDOWN (rs == SHUTDOWN && firstTask == null)) { // 等于SHUTDOWN并且firstTask为null if (t.isAlive()) // precheck that t is startable // 线程刚添加进来，还未启动就存活 // 抛出线程状态异常 throw new IllegalThreadStateException(); // 将worker添加到worker集合 workers.add(w); // 获取worker集合的大小 int s = workers.size(); if (s > largestPoolSize) // 队列大小大于largestPoolSize // 重新设置largestPoolSize largestPoolSize = s; // 设置worker已被添加标识 workerAdded = true; } } finally { // 释放锁 mainLock.unlock(); } if (workerAdded) { // worker被添加 // 开始执行worker的run方法 t.start(); // 设置worker已开始标识 workerStarted = true; } } } finally { if (! workerStarted) // worker没有开始 // 添加worker失败 addWorkerFailed(w); } return workerStarted; } ### 执行任务 ### runWorker函数中会实际执行给定任务（即调用用户重写的run方法），并且当给定任务完成后，会继续从阻塞队列中取任务，直到阻塞队列为空（即任务全部完成）。在执行给定任务时，会调用钩子函数，利用钩子函数可以完成用户自定义的一些逻辑。在runWorker中会调用到getTask函数和processWorkerExit钩子函数 final void runWorker(Worker w) { // 获取当前线程 Thread wt = Thread.currentThread(); // 获取w的firstTask Runnable task = w.firstTask; // 设置w的firstTask为null w.firstTask = null; // 释放锁（设置state为0，允许中断） w.unlock(); // allow interrupts boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { // 任务不为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) || // 线程池的运行状态至少应该高于STOP (Thread.interrupted() && // 线程被中断 runStateAtLeast(ctl.get(), STOP))) && // 再次检查，线程池的运行状态至少应该高于STOP !wt.isInterrupted()) // wt线程（当前线程）没有被中断 wt.interrupt(); // 中断wt线程（当前线程） 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; // 增加给worker完成的任务数量 w.completedTasks++; // 释放锁 w.unlock(); } } completedAbruptly = false; } finally { // 处理完成后，调用钩子函数 processWorkerExit(w, completedAbruptly); } } 此函数用于从workerQueue阻塞队列中获取Runnable对象，由于是阻塞队列，所以支持有限时间等待（poll）和无限时间等待（take）。在该函数中还会响应shutDown和、shutDownNow函数的操作，若检测到线程池处于SHUTDOWN或STOP状态，则会返回null，而不再返回阻塞队列中的Runnalbe对象。 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())) { // 大于等于SHUTDOWN（表示调用了shutDown）并且（大于等于STOP（调用了shutDownNow）或者worker阻塞队列为空） // 减少worker的数量 decrementWorkerCount(); // 返回null，不执行任务 return null; } // 获取worker数量 int wc = workerCountOf(c); // Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 是否允许coreThread超时或者workerCount大于核心大小 if ((wc > maximumPoolSize || (timed && timedOut)) // worker数量大于maximumPoolSize && (wc > 1 || workQueue.isEmpty())) { // workerCount大于1或者worker阻塞队列为空（在阻塞队列不为空时，需要保证至少有一个wc） if (compareAndDecrementWorkerCount(c)) // 比较并减少workerCount // 返回null，不执行任务，该worker会退出 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; } } } processWorkerExit函数是在worker退出时调用到的钩子函数，而引起worker退出的主要因素如下 1. 阻塞队列已经为空，即没有任务可以运行了。 2. 调用了shutDown或shutDownNow函数 此函数会根据是否中断了空闲线程来确定是否减少workerCount的值，并且将worker从workers集合中移除并且会尝试终止线程池。 private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // 如果被中断，则需要减少workCount // If abrupt, then workerCount wasn't adjusted decrementWorkerCount(); // 获取可重入锁 final ReentrantLock mainLock = this.mainLock; // 获取锁 mainLock.lock(); try { // 将worker完成的任务添加到总的完成任务中 completedTaskCount += w.completedTasks; // 从workers集合中移除该worker workers.remove(w); } finally { // 释放锁 mainLock.unlock(); } // 尝试终止 tryTerminate(); // 获取线程池控制状态 int c = ctl.get(); if (runStateLessThan(c, STOP)) { // 小于STOP的运行状态 if (!completedAbruptly) { int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) // 允许核心超时并且workQueue阻塞队列不为空 min = 1; if (workerCountOf(c) >= min) // workerCount大于等于min // 直接返回 return; // replacement not needed } // 添加worker addWorker(null, false); } } ### 关闭线程池 ### public void shutdown() { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 检查shutdown权限 checkShutdownAccess(); // 设置线程池控制状态为SHUTDOWN advanceRunState(SHUTDOWN); // 中断空闲worker interruptIdleWorkers(); // 调用shutdown钩子函数 onShutdown(); // hook for ScheduledThreadPoolExecutor } finally { mainLock.unlock(); } // 尝试终止 tryTerminate(); } final void tryTerminate() { for (;;) { // 无限循环，确保操作成功 // 获取线程池控制状态 int c = ctl.get(); if (isRunning(c) || // 线程池的运行状态为RUNNING runStateAtLeast(c, TIDYING) || // 线程池的运行状态最小要大于TIDYING (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) // 线程池的运行状态为SHUTDOWN并且workQueue队列不为null // 不能终止，直接返回 return; if (workerCountOf(c) != 0) { // 线程池正在运行的worker数量不为0 // Eligible to terminate // 仅仅中断一个空闲的worker interruptIdleWorkers(ONLY_ONE); return; } // 获取线程池的锁 final ReentrantLock mainLock = this.mainLock; // 获取锁 mainLock.lock(); try { if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { // 比较并设置线程池控制状态为TIDYING try { // 终止，钩子函数 terminated(); } finally { // 设置线程池控制状态为TERMINATED ctl.set(ctlOf(TERMINATED, 0)); // 释放在termination条件上等待的所有线程 termination.signalAll(); } return; } } finally { // 释放锁 mainLock.unlock(); } // else retry on failed CAS } } private void interruptIdleWorkers(boolean onlyOne) { // 线程池的锁 final ReentrantLock mainLock = this.mainLock; // 获取锁 mainLock.lock(); try { for (Worker w : workers) { // 遍历workers队列 // worker对应的线程 Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) { // 线程未被中断并且成功获得锁 try { // 中断线程 t.interrupt(); } catch (SecurityException ignore) { } finally { // 释放锁 w.unlock(); } } if (onlyOne) // 若只中断一个，则跳出循环 break; } } finally { // 释放锁 mainLock.unlock(); } } ### 例子 ### import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; public class Demo { public static void main(String[] args) { // 10个线程来处理大量的任务 // ThreadPoolExecutor pool = new ThreadPoolExecutor(10, 10, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()); ExecutorService pool = Executors.newFixedThreadPool(10); // ExecutorService pool = Executors.newCachedThreadPool(); // ExecutorService pool = Executors.newSingleThreadExecutor(); // ScheduledExecutorService pool = Executors.newScheduledThreadPool(10); // ExecutorService pool = Executors.newWorkStealingPool(); while(true) { Future<?> f = pool.submit(new Runnable() { @Override public void run() { } }); // pool.schedule(new Runnable() { // // @Override // public void run() { // System.out.println(Thread.currentThread().getName()); // } // }, 5, TimeUnit.SECONDS); // pool.execute(new Runnable() { // // @Override // public void run() { // System.out.println(Thread.currentThread().getName()); // try { // Thread.sleep(100); // } catch (InterruptedException e) { // e.printStackTrace(); // } // } // }); } } } 来自龙果学院讲师：叶子猿老师
相关 Java 线程池、Runnable线程池、Callable线程池 线程池： 其实就是一个容纳多个线程的容器，其中的线程可以反复的使用，省去了频繁创建和销毁过程对象的操作，无需反复创建线程面消耗过多资源。 为什么要用线程池： 合理 青旅半醒/ 2023年02月26日 12:30/ 0 赞/ 27 阅读
相关 线程、线程池 创建线程的3种方法： package com.frank.threadPool.createThread; / @author 小石潭记 布满荆棘的人生/ 2022年10月22日 04:27/ 0 赞/ 364 阅读
相关 线程池 1.所谓线程池，就是程序的初始化阶段，就预先创建一批线程，每个线程都做好准备干活； 2.然后有一个任务列表，一开始为空，当有任务来了，就往任务列表里面添加；这个任务列表 痛定思痛。/ 2022年06月13日 13:22/ 0 赞/ 309 阅读
相关 线程池 西施越溪女，明艳光云海 最近用线程池和不用线程池做了个速度的测试，在这里备注下： 结果是速度不相上下； public static void main(Str 妖狐艹你老母/ 2022年05月20日 02:35/ 0 赞/ 262 阅读
相关 线程池 线程池 Java里面线程池的顶级接口是 java.util.concurrent.Executor ， 但是严格意义上讲 Executor并不是一个线程池，而只是一个 迈不过友情╰/ 2022年03月06日 14:34/ 0 赞/ 381 阅读
相关 线程池 线程池 > 从字面义上来讲，是指管理一组同构工作线程的资源池。线程池是与工作队列密切相关的，其中在工作队列中(Worker Queue)保存了所有等待执行的任务。工作者( 清疚/ 2021年12月11日 03:35/ 0 赞/ 365 阅读
相关 线程池 可preStart一个或全部core thread 0，小于core则来一个任务建一个线程（firstTask），队列，额外线程，拒绝 一个AtomicInteger的 今天药忘吃喽~/ 2021年11月23日 03:40/ 0 赞/ 398 阅读
相关 线程池 1、先创建线程池 import java.util.concurrent.ArrayBlockingQueue; import java.util.concu 拼搏现实的明天。/ 2021年11月09日 14:28/ 0 赞/ 406 阅读
青旅半醒
今天药忘吃喽~
还没有评论，来说两句吧...