并发编程——线程池 ╰+攻爆jí腚メ 2023-10-16 11:57 3阅读 0赞 #### 文章目录 #### * 线程池介绍 * 线程池核心属性 * 核心方法 * * execute方法 * addWorker添加工作线程 * runWorker执行任务 * getTask工作线程排队拿任务 ## 线程池介绍 ## Java构建线程的方式 * new Thread * new Runnable * new Callable 为了避免频繁创建和销毁线程造成不必要的性能，一般在使用线程时，会采用线程池。 线程池使用方式： public static void main(String[] args) { // 线程池的核心线程数如何设置 // 任务可以分为两种：CPU密集，IO密集。 ThreadPoolExecutor executor = new ThreadPoolExecutor( 1, 2, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1), new ThreadFactory() { @Override public Thread newThread(Runnable r) { Thread t = new Thread(r); // ... return t; } }, new ThreadPoolExecutor.AbortPolicy() ); executor.execute(任务); executor.submit(有返回结果的任务); } 线程池提供的拒绝策略： * AbortPolicy：抛出异常。 * CallerRunsPolicy：让提交任务的线程处理这个任务。 * DiscardPolicy：丢弃任务 * DiscardOldestPolicy：扔掉队列最前面的任务，尝试把当前任务添加进去。 ## 线程池核心属性 ## // AtomicInteger,就是一个int，写操作用CAS实现，保证了原子性 // ctl维护线程池的2个核心内容： // 1：线程池状态（高3位，维护着线程池状态） // 2：工作线程数量（核心线程+非核心线程，低29位，维护着工作线程个数） private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // COUNT_BITS=29 private static final int COUNT_BITS = Integer.SIZE - 3; // 工作线程的最大个数 // 00100000 00000000 00000000 00000000 - 1 // 000111111111111111111111111111111 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; 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; } 线程池状态 1. RUNNING：正常运行的状态，接受新的任务，处理等待队列中的任务。线程池的初始化状态是RUNNING。一旦线程池被创建，就处于RUNNING状态，并且线程池中的任务数为0。 2. SHUTDOWN：不接受新的任务提交，但是会继续处理等待队列中的任务。调用线程池的shutdown()方法时，线程池由RUNNING变为SHUTDOWN。 3. STOP：不接受新的任务提交，不再处理等待队列中的任务，中断正在执行任务的线程。调用线程池的shutdownNow()方法时，线程池由(RUNNING or SHUTDOWN)变为STOP。 4. TIDYING：所有的任务都销毁了，workCount为0，线程池的状态在转换为TIDYING状态时，会执行钩子方法terminated()。 5. TERMINATED：线程池彻底终止。 ## 核心方法 ## ### execute方法 ### 通过execute方法，可以看到线程池处理任务的整体执行流程 public void execute(Runnable command) { if (command == null) throw new NullPointerException(); // 拿到ctl int c = ctl.get(); // 通过ctl获取当前工作线程个数 if (workerCountOf(c) < corePoolSize) { // true：代表是核心线程，false：代表是非核心线程 if (addWorker(command, true)) // 如果添加核心线程成功，return结束掉 return; // 如果添加失败，重新获取ctl c = ctl.get(); } // 核心线程数已经到了最大值、添加时，线程池状态变为SHUTDOWN/STOP // 判断线程池是否是运行状态 && 添加任务到工作队列 if (isRunning(c) && workQueue.offer(command)) { // 再次获取ctl的值 int recheck = ctl.get(); // 再次判断线程池状态。 DCL // 如果状态不是RUNNING，把任务从工作队列移除。 if (! isRunning(recheck) && remove(command)) // 走一波拒绝策略。 reject(command); // 线程池状态是RUNNING。 // 判断工作线程数是否是0个。 // 可以将核心线程设置为0，所有工作线程都是非核心线程。 // 核心线程也可以通过keepAlived超时被销毁，所以如果恰巧核心线程被销毁，也会出现当前效果 else if (workerCountOf(recheck) == 0) // 添加空任务的非核心线程去处理工作队列中的任务 addWorker(null, false); } // 可能工作队列中的任务存满了，没添加进去，到这就要添加非核心线程去处理任务 else if (!addWorker(command, false)) // 执行拒绝策略！ reject(command); } 这个方法的主要逻辑是： 1. 如果当前运行的线程数少于corePoolSize，尝试启动一个新线程并将给定的任务作为它的第一个任务。这个调用会原子性地检查runState和workerCount，以防止在不应该添加线程时发出错误警报。 2. 如果任务可以成功排队，那么我们仍然需要再次检查是否应该添加一个新线程（因为自上次检查以来，现有的线程可能已经死亡），或者自进入此方法以来，线程池是否已经关闭。所以我们重新检查状态，并在必要时回滚排队，如果已停止，则启动一个新线程。 3. 如果我们无法排队任务，那么我们尝试添加一个新线程。如果失败，我们知道已经关闭或饱和，所以拒绝任务。 ### addWorker添加工作线程 ### private boolean addWorker(Runnable firstTask, boolean core) { xxx: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 判断线程池状态 if (rs >= SHUTDOWN && // 判断如果线程池的状态为SHUTDOWN，还要处理工作队列中的任务 // 如果你添加工作线程的方式，是任务的非核心线程，并且工作队列还有任务 ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; // 判断工作线程个数 for (;;) { int wc = workerCountOf(c); // 判断1：工作线程是否已经 == 工作线程最大个数 // 判断2-true判断：判断是核心线程么？如果是判断是否超过核心线程个数 // 判断2-false判断：如果是非核心线程，查看是否超过设置的最大线程数 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 对工作线程进行 + 1操作 if (compareAndIncrementWorkerCount(c)) // +1成功，跳出外层循环，执行添加工作线程的业务 // 以CAS方式，对ctl+1，多线程并发操作，只有会有一个成功 break xxx; // 重新拿ctl， c = ctl.get(); // 判断线程池状态是否有变化 if (runStateOf(c) != rs) continue xxx; } } // 添加工作线程的业务 boolean workerStarted = false; boolean workerAdded = false; // Worker就是工作线程 Worker w = null; try { // 创建工作线程，将任务传到Worker中 w = new Worker(firstTask); final Thread t = w.thread; // 只有你写的线程工厂返回的是null，这里才会为null if (t != null) { // 获取锁资源 final ReentrantLock mainLock = this.mainLock; // 加锁。 因为我要在启动这个工作线程时，避免线程池状态发生变化，加锁。 mainLock.lock(); try { // 重新获取ctl，拿到线程池状态 int rs = runStateOf(ctl.get()); // DCL i think you know~~~ if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { // 判断Worker中的thread是否已经启动了，一般不会启动，除非你在线程工厂把他启动了 if (t.isAlive()) throw new IllegalThreadStateException(); // 将工作线程存储到hashSet中 workers.add(w); // 获取工作线程个数，判断是否需要修改最大工作线程数记录。 int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; // 工作线程添加成功 0 workerAdded = true; } } finally { mainLock.unlock(); } // 如果添加成功 if (workerAdded) { // 启动工作线程 t.start(); // 设置标识为true workerStarted = true; } } } finally { // 如果工作线程启动失败 if (! workerStarted) addWorkerFailed(w); } return workerStarted; } // 如果添加工作线程失败，执行 private void addWorkerFailed(Worker w) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 说明worker可能存放到了workers的hashSet中。 if (w != null) // 移除！ workers.remove(w); // 减掉workerCount的数值 -1 decrementWorkerCount(); // 尝试干掉自己 tryTerminate(); } finally { mainLock.unlock(); } } 这个方法的主要逻辑是： 1. 检查是否正在关闭或已经关闭，如果是，根据是否有任务以及队列是否为空来决定是否返回false。 2. 如果没有关闭，那么检查当前线程数是否超过了核心线程数或最大线程数，如果超过了则返回false。 3. 如果没有超过线程数限制，那么通过CAS操作增加workerCount，如果成功则跳出循环。如果失败，则重新读取ctl并重新计算runState，如果runState发生变化则继续循环。如果未发生变化，则继续尝试增加workerCount。 4. 创建Worker对象，并启动对应的线程。如果启动成功，则返回true。如果启动失败，则调用addWorkerFailed()方法处理失败情况。 ### runWorker执行任务 ### final void runWorker(Worker w) { // 拿到当前线程对象 Thread wt = Thread.currentThread(); // 拿到worker中存放的Runnable Runnable task = w.firstTask; // 将worker中的任务清空 w.firstTask = null; boolean completedAbruptly = true; try { // 如果Worker自身携带任务，直接执行 // 如果Worker携带的是null，通过getTask去工作队列获取任务 while (task != null || (task = getTask()) != null) { w.lock(); // 判断线程池状态是否大于等于STOP，如果是要中断当前线程 if ((runStateAtLeast(ctl.get(), STOP) || // 中断当前线程（DCL） (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); } } 这个方法的主要逻辑是： 1. 获取当前线程和任务。如果任务不为空，则执行任务。如果任务为空，则从任务队列中获取任务。如果获取任务失败，则退出循环。 2. 在执行任务前，检查线程是否中断，如果中断则停止执行任务。如果未中断，则执行任务。如果抛出运行时异常或错误，则捕获并记录异常。如果抛出其他异常，则转换为错误并抛出。 3. 在执行任务后，将任务设置为null，增加已完成任务数，并解锁worker。 4. 如果循环结束时，任务为空，则说明任务队列已关闭或已空，将completedAbruptly设置为false。否则，说明任务队列中还有任务，将completedAbruptly保持为true。 5. 最后，处理worker退出，根据completedAbruptly的值决定是否重置线程池状态。 ### getTask工作线程排队拿任务 ### private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); // 线程池状态判断 // 如果线程池状态为SHUTDOWN && 工作队列为空 // 如果线程池状态为STOP if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { // 对工作线程个数-- decrementWorkerCount(); return null; } // 对数量的判断。 int wc = workerCountOf(c); // 判断核心线程是否允许超时？ // 工作线程个数是否大于核心线程数 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 判断工作线程是否超过了最大线程数 && 工作队列为null if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { // 工作线程数有问题，必须-1，干掉当前工作线程 // 工作线程是否超过了核心线程，如果超时，就干掉当前线程 // 对工作线程个数-- if (compareAndDecrementWorkerCount(c)) return null; continue; } try { // 如果是非核心，走poll，拉取工作队列任务， // 如果是核心线程，走take一直阻塞，拉取工作队列任务 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : // 当工作队列没有任务时，这时就会被Condition通过await阻塞线程 // 当有任务添加到工作线程后，这是添加完任务后，就会用过Condition.signal唤醒阻塞的线程 workQueue.take(); if (r != null) return r; // 执行的poll方法，并且在指定时间没拿到任务， timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } } 该方法的主要逻辑是： 1. 检查线程池状态，如果线程池已经关闭或停止，并且任务队列为空，则减少worker数量并返回null。 2. 检查worker数量是否超过了最大线程数或核心线程数，如果是，则减少worker数量并返回null。 3. 如果worker数量未超过最大线程数或核心线程数，则从任务队列中获取任务。如果获取任务失败，则返回null。如果获取任务成功，则返回任务。 4. 如果在获取任务时发生中断，则重新尝试获取任务。如果在指定的时间内无法获取任务，则将timedOut标志设置为true。如果在指定的时间内无法获取任务并且允许超时，则减少worker数量并返回null。
还没有评论，来说两句吧...