| 线程池(ThreadPoolExecutor)源码分析之如何保证核心线程不被销毁的 | 您所在的位置：网站首页 › 线程池如何销毁线程 › 线程池(ThreadPoolExecutor)源码分析之如何保证核心线程不被销毁的 |
线程池(ThreadPoolExecutor)源码分析之如何保证核心线程不被销毁的
|
今天看到了别人的一个代码，为了实现每小时重启一下MQ拉取消息，他使用的是Thread.sleep（10006060）方法，然后重启MQ。我一看到就非常头疼啊。。为什么要使用这种方式而不使用java的线程池呢？于是我就问他，他说当时为了方便。大家都知道Thread.sleep期间是不会释放共享资源的，会造成死锁现象。然后我就想Thread.sleep可以在睡觉过程中等待被interrupt中断，然后继续工作。那么线程池是怎么保证他的核心线程不释放 而一直等待任务的执行的呢？难道我们一直理解的线程run方法执行完毕线程就销毁是不正确的？而且还有我们为何通过设置allowCoreThreadTimeOut(true) 就能使核心线程销毁的呢？ 我们通常都是通过执行execute(Runnable command)方法来向线程池提交一个不需要返回结果的任务的（如果你需要返回结果那么就是 Future submit(Callable task)方法）,怀着一颗探索的心，敲敲翻开了线程池的源码： public void execute(Runnable command) { /*如果提交的任务为null 抛出空指针异常*/ if (command == null) throw new NullPointerException(); int c = ctl.get(); /*如果当前的任务数小于等于设置的核心线程大小，那么调用addWorker直接执行该任务*/ if (workerCountOf(c) int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } /*如果向队列中添加失败，那么就新开启一个线程来执行该任务*/ else if (!addWorker(command, false)) reject(command); }有人可能不理解ctl.get()简单解释一下，线程池是通过Integer类型的高3为表述当前线程池的状态RUNNING，SHUTDOWN，STOP，TIDYING，TERMINATED 。低29位表示当前线程的运行任务数量。然后通过位运算来计算运行状态和任务的数量。 解释一下execute的执行流程 当线程池中的线程数小于corePoolSize 时，新提交的任务直接新建一个线程执行任务（不管是否有空闲线程） 当线程池中的线程数等于corePoolSize 时，新提交的任务将会进入阻塞队列（workQueue）中，等待线程的调度 当阻塞队列满了以后，如果corePoolSize < maximumPoolSize ,则新提交的任务会新建线程执行任务，直至线程数达到maximumPoolSize 当线程数达到maximumPoolSize 时，新提交的任务会由(饱和策略)管理 代码执行逻辑很简单，我们需要注意的就是addWorker方法，点进去继续查看。 private boolean addWorker(Runnable firstTask, boolean core) { retry: //死循环 for (;;) { int c = ctl.get(); int rs = runStateOf(c); //如果当前线程池的状态时SHUTDOWN，STOP，TIDYING，TERMINATED并且为SHUTDOWN状态时任务队列为空，那么就返回false 原因：如果调用了shutdown方法，此时的线程池还会继续工作并且会在任务队列中的所有任务执行完成后才会结束线程池。 // Check if queue empty only if necessary. if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; //死循环 for (;;) { int wc = workerCountOf(c); //core是在execute方法中传的参数，true表示 核心线程，false表示最大线程 //CAPACITY 可以理解为Integer的最大值 1左移29位再-1 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 //finally块释放内置锁 mainLock.unlock(); } //如果任务添加成功那么开始执行任务 if (workerAdded) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }写到这里差点忘记了我们当初的目的，我们的目的是查看线程池是如何保证核心线程不被销毁的。 看到这里终于出现了一点眉目，在任务添加成功后，我们发现了t.start() if (workerAdded) { t.start(); workerStarted = true; }可不要被迷惑，这里的t是通过work.thread; 得到的。这时候我们需要查看work类中的run方法。 work在ThreadPoolExecutor为一个内部类实现了Runnable接口。只有一个构造方法 Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** 重写了run方法 */ public void run() { runWorker(this); }通过构造方法我们可以清楚的看到我们提交的任务就是firstTask，而thread就是当前的work对象。在上面的addWorker方法中调用的t.start()就会调用这里的runWorker(this)方法，点进去继续查看 final void runWorker(Worker w) { Thread wt = Thread.currentThread(); //得到worker对象中我们提交的任务 Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //如果当前任务为空 那么就从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 { 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循环。如果task不为空执行task否则从getTask()中取任务。在执行完任务后会在finally 块中设置task = null; 顺便介绍一下，在这里，我们可以看到beforeExecute(Thread t, Runnable r)方法和afterExecute(Runnable r, Throwable t)会在任务的执行前后执行，我们可以通过继承线程池的方式来重写这两个方法，这样就能够对任务的执行进行监控啦。 咋一看 好像没什么问题。其实我们可以发现如果执行完一个任务 task 设置为 null。就要调用 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? 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; } } }需要仔细关注的是这些代码 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r;解释一下：从阻塞任务队列中取任务，如果设置了allowCoreThreadTimeOut(true) 或者当前运行的任务数大于设置的核心线程数，那么timed =true 。此时将使用workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)从任务队列中取任务，而如果没有设置，那么使用workQueue.take();取任务，对于阻塞队列，poll(long timeout, TimeUnit unit) 将会在规定的时间内去任务，如果没取到就返回null。take()会一直阻塞，等待任务的添加。 到此 相信我们都能够理解为什么我们的线程池能够一直等待任务的执行而不被销毁了，其实也就是进入了阻塞状态而已。 关注我，随时获取最新文章 |
【本文地址】