shutdown()和shutdownNow()

shutdown()

shutdown()会尝试中断空闲的线程，并把ThreadPool的状态置成SHUTDOWN,
这个状态下，不能通过submit()或者execute()提交任务，但是正在执行的和任务队列中的任务还可以继续执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

任务队列中的任务拉取, 通过getTask()方法进行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
final void runWorker(Worker w) {
      Thread wt = Thread.currentThread();
      Runnable task = w.firstTask;
      w.firstTask = null;
      w.unlock(); // allow interrupts
      boolean completedAbruptly = true;
      try {
          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);
      }
  }

shutdownNow()

shutdownNow()会中断所有的worker线程， 然后将队列里没有执行的任务全部停止并返回。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public List<Runnable> shutdownNow() {
       List<Runnable> tasks;
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
           checkShutdownAccess();
           advanceRunState(STOP);
           interruptWorkers();
           tasks = drainQueue();
       } finally {
           mainLock.unlock();
       }
       tryTerminate();
       return tasks;
   }

中断线程池中的线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
    * Interrupts all threads, even if active. Ignores SecurityExceptions
    * (in which case some threads may remain uninterrupted).
    */
   private void interruptWorkers() {
       final ReentrantLock mainLock = this.mainLock;
       mainLock.lock();
       try {
           for (Worker w : workers)
               w.interruptIfStarted();
       } finally {
           mainLock.unlock();
       }
   }

    void interruptIfStarted() {
           Thread t;
           if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
               try {
                   t.interrupt();
               } catch (SecurityException ignore) {
               }
           }
       }

停止任务队列中没有执行的任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/**
    * Drains the task queue into a new list, normally using
    * drainTo. But if the queue is a DelayQueue or any other kind of
    * queue for which poll or drainTo may fail to remove some
    * elements, it deletes them one by one.
    */
   private List<Runnable> drainQueue() {
       BlockingQueue<Runnable> q = workQueue;
       ArrayList<Runnable> taskList = new ArrayList<Runnable>();
       q.drainTo(taskList);
       if (!q.isEmpty()) {
           for (Runnable r : q.toArray(new Runnable[0])) {
               if (q.remove(r))
                   taskList.add(r);
           }
       }
       return taskList;
   }

java中线程的停止是协作式的，线程要主动检查interrupted状态，并做出响应才能正确退出。

JVM退出

  • 非守护线程

如果线程池里的线程是非守护线程， JVM会等待线程的退出，然后才会真正的推出。

  • 守护线程（不推荐）

JVM退出的时候不会等待守护线程，因此如果要等待所有的任务执行完后再退出，需要自己代码处理, 可以使用awaitTermination方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (;;) {
                if (runStateAtLeast(ctl.get(), TERMINATED))
                    return true;
                if (nanos <= 0)
                    return false;
                nanos = termination.awaitNanos(nanos);
            }
        } finally {
            mainLock.unlock();
        }
    }

其中termination是结束的条件队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

/**
 * Lock held on access to workers set and related bookkeeping.
 * While we could use a concurrent set of some sort, it turns out
 * to be generally preferable to use a lock. Among the reasons is
 * that this serializes interruptIdleWorkers, which avoids
 * unnecessary interrupt storms, especially during shutdown.
 * Otherwise exiting threads would concurrently interrupt those
 * that have not yet interrupted. It also simplifies some of the
 * associated statistics bookkeeping of largestPoolSize etc. We
 * also hold mainLock on shutdown and shutdownNow, for the sake of
 * ensuring workers set is stable while separately checking
 * permission to interrupt and actually interrupting.
 */
private final ReentrantLock mainLock = new ReentrantLock();

/**
 * Wait condition to support awaitTermination
 */
private final Condition termination = mainLock.newCondition();

线程池在退出的时候会唤醒等待在这个队列上的线程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
                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
        }
    }

线程池状态:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
    * The run state of this task, initially NEW.  The run state
    * transitions to a terminal state only in methods set,
    * setException, and cancel.  During completion, state may take on
    * transient values of COMPLETING (while outcome is being set) or
    * INTERRUPTING (only while interrupting the runner to satisfy a
    * cancel(true)). Transitions from these intermediate to final
    * states use cheaper ordered/lazy writes because values are unique
    * and cannot be further modified.
    *
    * Possible state transitions:
    * NEW -> COMPLETING -> NORMAL
    * NEW -> COMPLETING -> EXCEPTIONAL
    * NEW -> CANCELLED
    * NEW -> INTERRUPTING -> INTERRUPTED
    */
   private volatile int state;
   private static final int NEW          = 0;
   private static final int COMPLETING   = 1;
   private static final int NORMAL       = 2;
   private static final int EXCEPTIONAL  = 3;
   private static final int CANCELLED    = 4;
   private static final int INTERRUPTING = 5;
   private static final int INTERRUPTED  = 6;

参考

  1. 拆轮子：全面剖析 ThreadPoolExecutor（2）-众人拾柴 - 简书