|
@@ -236,7 +236,7 @@ private final class Worker extends AbstractQueuedSynchronizer implements Runnabl
|
|
|
}
|
|
|
```
|
|
|
|
|
|
-#### runWorker方法
|
|
|
+### runWorker方法
|
|
|
|
|
|
```Java
|
|
|
final void runWorker(Worker w) {
|
|
@@ -282,7 +282,7 @@ final void runWorker(Worker w) {
|
|
|
}
|
|
|
```
|
|
|
|
|
|
-#### 从任务队列中取出一个任务
|
|
|
+### 从任务队列中取出一个任务
|
|
|
|
|
|
```Java
|
|
|
private Runnable getTask() {
|
|
@@ -336,7 +336,7 @@ private Runnable getTask() {
|
|
|
> 3. 线程池线程数大于最大线程数
|
|
|
> 4. 线程可以被超时回收的情况下等待新任务超时
|
|
|
|
|
|
-#### 工作线程退出
|
|
|
+### 工作线程退出
|
|
|
|
|
|
```Java
|
|
|
private void processWorkerExit(Worker w, boolean completedAbruptly) {
|
|
@@ -377,3 +377,261 @@ private void processWorkerExit(Worker w, boolean completedAbruptly) {
|
|
|
}
|
|
|
}
|
|
|
```
|
|
|
+
|
|
|
+### 工作线程如何回收
|
|
|
+
|
|
|
+```java
|
|
|
+ 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 {
|
|
|
+ // 通过时间进行判断,如果超时返回false
|
|
|
+ Runnable r = timed ?
|
|
|
+ workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
|
|
|
+ workQueue.take();
|
|
|
+ if (r != null)
|
|
|
+ return r;
|
|
|
+ timedOut = true;
|
|
|
+ } catch (InterruptedException retry) {
|
|
|
+ timedOut = false;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+```
|
|
|
+
|
|
|
+```java
|
|
|
+private void processWorkerExit(Worker w, boolean completedAbruptly) {
|
|
|
+ // 如果completedAbruptly为true则表示任务执行过程中抛出了未处理的异常
|
|
|
+ // 所以还没有正确地减少worker计数,这里需要减少一次worker计数
|
|
|
+ if (completedAbruptly)
|
|
|
+ decrementWorkerCount();
|
|
|
+
|
|
|
+ final ReentrantLock mainLock = this.mainLock;
|
|
|
+ mainLock.lock();
|
|
|
+ try {
|
|
|
+ // 把将被销毁的线程已完成的任务数累加到线程池的完成任务总数上
|
|
|
+ completedTaskCount += w.completedTasks;
|
|
|
+ // 这段代码就会减少工作线程
|
|
|
+ workers.remove(w); // 从工作线程集合中移除该工作线程
|
|
|
+ } finally {
|
|
|
+ mainLock.unlock();
|
|
|
+ }
|
|
|
+
|
|
|
+ // 尝试结束线程池
|
|
|
+ tryTerminate();
|
|
|
+
|
|
|
+ int c = ctl.get();
|
|
|
+ // 如果是RUNNING 或 SHUTDOWN状态
|
|
|
+ if (runStateLessThan(c, STOP)) {
|
|
|
+ // worker是正常执行完
|
|
|
+ if (!completedAbruptly) {
|
|
|
+ // 如果允许核心线程超时则最小线程数是0,否则最小线程数等于核心线程数
|
|
|
+ int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
|
|
|
+ // 如果阻塞队列非空,则至少要有一个线程继续执行剩下的任务
|
|
|
+ if (min == 0 && ! workQueue.isEmpty())
|
|
|
+ min = 1;
|
|
|
+ // 如果当前线程数已经满足最小线程数要求,则不需要再创建替代线程
|
|
|
+ if (workerCountOf(c) >= min)
|
|
|
+ return; // replacement not needed
|
|
|
+ }
|
|
|
+ // 重新创建一个worker来代替被销毁的线程
|
|
|
+ addWorker(null, false);
|
|
|
+ }
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+## 如何中断
|
|
|
+
|
|
|
+### 中断别人
|
|
|
+
|
|
|
+```java
|
|
|
+private void processWorkerExit(Worker w, boolean completedAbruptly) {
|
|
|
+ // 尝试结束线程池
|
|
|
+ // 这里会中断别人
|
|
|
+ tryTerminate();
|
|
|
+}
|
|
|
+```
|
|
|
+
|
|
|
+### 具体实现
|
|
|
+
|
|
|
+```java
|
|
|
+ final void tryTerminate() {
|
|
|
+ for (;;) {
|
|
|
+ int c = ctl.get();
|
|
|
+ if (isRunning(c) ||
|
|
|
+ runStateAtLeast(c, TIDYING) ||
|
|
|
+ (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
|
|
|
+ return;
|
|
|
+ if (workerCountOf(c) != 0) { // Eligible to terminate
|
|
|
+ // 中断 ONLY_ONE = false
|
|
|
+ interruptIdleWorkers(ONLY_ONE);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ final ReentrantLock mainLock = this.mainLock;
|
|
|
+ mainLock.lock();
|
|
|
+ try {
|
|
|
+ if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
|
|
|
+ try {
|
|
|
+ terminated();
|
|
|
+ } finally {
|
|
|
+ ctl.set(ctlOf(TERMINATED, 0));
|
|
|
+ termination.signalAll();
|
|
|
+ }
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ } finally {
|
|
|
+ mainLock.unlock();
|
|
|
+ }
|
|
|
+ // else retry on failed CAS
|
|
|
+ }
|
|
|
+ }
|
|
|
+```
|
|
|
+
|
|
|
+```java
|
|
|
+ private void interruptIdleWorkers(boolean onlyOne) {
|
|
|
+ final ReentrantLock mainLock = this.mainLock;
|
|
|
+ mainLock.lock();
|
|
|
+ try {
|
|
|
+ for (Worker w : workers) {
|
|
|
+ Thread t = w.thread;
|
|
|
+ if (!t.isInterrupted() && w.tryLock()) {
|
|
|
+ try {
|
|
|
+ // 前面已经被中断了,中断别人,相当于链式的中断
|
|
|
+ t.interrupt();
|
|
|
+ } catch (SecurityException ignore) {
|
|
|
+ } finally {
|
|
|
+ w.unlock();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (onlyOne)
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ } finally {
|
|
|
+ mainLock.unlock();
|
|
|
+ }
|
|
|
+ }
|
|
|
+```
|
|
|
+
|
|
|
+### 唤醒
|
|
|
+
|
|
|
+```java
|
|
|
+ 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;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+```
|
|
|
+
|
|
|
+### 如果直接使用shutdown函数
|
|
|
+
|
|
|
+> 如果线程被中断会立即停止该线程在执行的任务吗?
|
|
|
+>
|
|
|
+> interrupt() 方法并不像在 for 循环语句中使用 break 语句那样干脆,马上就停止循环。调用 interrupt() 方法仅仅是在当前线程中打一个停止的标记,并不是真的停止线程。
|
|
|
+
|
|
|
+1. Running 状态,任务已全部完成,线程在阻塞等待。在 Running 状态进入 shutdown 状态。
|
|
|
+
|
|
|
+很简单,中断信号将其唤醒,从而进入下一轮循环。到达条件1处,符合条件,减少工作线程数量,并返回null,由外层结束这条线程。这里的decrementWorkerCount()是自旋式的,一定会减1
|
|
|
+
|
|
|
+2. Running 状态,任务还没有完全执行完
|
|
|
+
|
|
|
+```java
|
|
|
+Runnable r = timed ?
|
|
|
+ // 内部的方法
|
|
|
+ workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
|
|
|
+ workQueue.take();
|
|
|
+```
|
|
|
+
|
|
|
+```java
|
|
|
+ public E take() throws InterruptedException {
|
|
|
+ E x;
|
|
|
+ int c = -1;
|
|
|
+ final AtomicInteger count = this.count;
|
|
|
+ final ReentrantLock takeLock = this.takeLock;
|
|
|
+ // 中断获取锁
|
|
|
+ takeLock.lockInterruptibly();
|
|
|
+ try {
|
|
|
+ while (count.get() == 0) {
|
|
|
+ notEmpty.await();
|
|
|
+ }
|
|
|
+ x = dequeue();
|
|
|
+ c = count.getAndDecrement();
|
|
|
+ if (c > 1)
|
|
|
+ notEmpty.signal();
|
|
|
+ } finally {
|
|
|
+ takeLock.unlock();
|
|
|
+ }
|
|
|
+ if (c == capacity)
|
|
|
+ signalNotFull();
|
|
|
+ return x;
|
|
|
+ }
|
|
|
+
|
|
|
+ public void lockInterruptibly() throws InterruptedException {
|
|
|
+ sync.acquireInterruptibly(1);
|
|
|
+ }
|
|
|
+
|
|
|
+ public final void acquireInterruptibly(int arg)
|
|
|
+ throws InterruptedException {
|
|
|
+ // 重置中断状态
|
|
|
+ if (Thread.interrupted())
|
|
|
+ throw new InterruptedException();
|
|
|
+ if (!tryAcquire(arg))
|
|
|
+ doAcquireInterruptibly(arg);
|
|
|
+ }
|
|
|
+```
|