时间:2022-02-24 10:45:47 | 栏目:Android代码 | 点击:次
今天面试被问到线程池如何复用线程的?当场就懵掉了...于是面试完毕就赶紧打开源码看了看,在此记录下:
我们都知道线程池的用法,一般就是先new一个ThreadPoolExecutor对象,再调用execute(Runnable runnable)传入我们的Runnable,剩下的交给线程池处理就行了,于是这次我就从ThreadPoolExecutor的execute方法看起:
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(); //1.如果workerCountOf(c)即正在运行的线程数小于核心线程数,就执行addWork if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } //2.如果线程池还在运行状态并且把任务添加到任务队列成功 if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); //3.如果线程池不在运行状态并且从任务队列移除任务成功,执行线程池饱和策略(默认直接抛出异常) if (! isRunning(recheck) && remove(command)) reject(command); //4.否则如果此时运行线程数==0,就直接调用addWork方法 else if (workerCountOf(recheck) == 0) addWorker(null, false); } //5.如果2条件不成立,继续判断如果addWork返回false,执行线程池饱和策略 else if (!addWorker(command, false)) reject(command); }
大致过程就是如果核心线程未满,则直接addWorker(该方法下面会再分析);如果核心线程已满,则尝试将任务加进消息队列中,并再判断如果此时运行线程数==0则调addWorker方法,否则不做任何处理(因为运行的线程处理完自己的任务后会去消息队列中取任务来执行,下面会分析);如果任务队列添加任务失败,那么直接addWorker(),如果addWorker返回false,执行饱和策略,下面我们就来看看addWorker里面做了什么
/** * @param firstTask the task the new thread should run first (or * null if none). Workers are created with an initial first task * (in method execute()) to bypass queuing when there are fewer * than corePoolSize threads (in which case we always start one), * or when the queue is full (in which case we must bypass queue). * Initially idle threads are usually created via * prestartCoreThread or to replace other dying workers. * * @param core if true use corePoolSize as bound, else * maximumPoolSize. (A boolean indicator is used here rather than a * value to ensure reads of fresh values after checking other pool * state). * @return true if successful */ 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 && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c); //1.如果正在运行的线程数大于corePoolSize 或 maximumPoolSize(core代表以核心线程数还是最大线程数为边界),return false,表示addWorker失败 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; //2.否则将运行线程数+1,并跳出这个for循环 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 { //3.创建一个Worker对象,传入我们的runnable w = new Worker(firstTask); final Thread t = w.thread; if (t != 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 || (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) { //4.开始启动线程 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker. */ public void run() { runWorker(this); } final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //1.当firstTask不为空或getTask不为空时一直循环 while (task != null || (task = getTask()) != 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) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { //2.执行任务 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); } }
可以看到addWorker方法主要就是先判断正在运行线程数是否超过了最大线程数(具体根据边界取),如果未超过则创建一个worker对象,其中firstTask是我们传入的Runnable,当然根据上面的execute方法可知当4条件满足时,传入的firstTask是null,Thread是用ThreadFactory创建的线程,传入的Runnable是Worker自己,最后开启线程,于是执行Worker这里的run、runWorker方法,在runWorker方法里,开启一个while循环,当firstTask不为空或getTask不为空时,执行task,下面我们接着看看getTask里面做了什么:
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? //1.会不会淘汰空闲线程 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; //2.return null意味着回收一个Worker即淘汰一个线程 if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { //3.等待指定时间 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
可以看1、2注释,allowCoreThreadTimeOut代表存活一定时间是否对核心线程有效(默认为false),先看它为ture的情况,此时不管是核心线程还是非核心线程在3处都会等待一定时间(就是我们传入的线程保活时间),等待时间内如果从任务队列取到任务,则返回执行,否则timeout为true,继续走到2,由于(timed && timedOut)和workQueue.isEmpty()均为true,返回null,代表回收一个线程;如果allowCoreThreadTimeOut为false,代表不回收核心线程,此时如果在3处没有取到任务,继续执行到2处,只有当wc > corePoolSize或wc > maximumPoolSize时才会执行return null,否则一直循环,相当于该线程一直处于运行状态,直到从任务队列拿到新的任务