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

/**

 * Initiates an orderly shutdown in which previously submitted

 * tasks are executed, but no new tasks will be accepted.

 * Invocation has no additional effect if already shut down.

 * 开始一个有序的关闭，在关闭中，之前提交的任务会被执行（包含正在执行的，在阻塞队列中的），但新任务会被拒绝

 * 如果线程池已经shutdown，调用此方法不会有附加效应

 *

 * <p>This method does not wait for previously submitted tasks to

 * complete execution.  Use {@link #awaitTermination awaitTermination}

 * to do that.

 * 当前方法不会等待之前提交的任务执行结束，可以使用awaitTermination()

 *

 * @throws SecurityException {@inheritDoc}

 */

public void shutdown() {

    final ReentrantLock mainLock = this.mainLock;

    mainLock.lock(); //上锁

    try {

        //判断调用者是否有权限shutdown线程池

        checkShutdownAccess();

        //CAS+循环设置线程池状态为shutdown

        advanceRunState(SHUTDOWN);

        //中断所有空闲线程

        interruptIdleWorkers();

        onShutdown(); // hook for ScheduledThreadPoolExecutor

    }

    finally {

        mainLock.unlock(); //解锁

    }

    //尝试终止线程池

    tryTerminate();

}

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

44

45

46

47

48

49

/**

 * Interrupts threads that might be waiting for tasks (as

 * indicated by not being locked) so they can check for

 * termination or configuration changes. Ignores

 * SecurityExceptions (in which case some threads may remain

 * uninterrupted).

 * 中断在等待任务的线程(没有上锁的)，中断唤醒后，可以判断线程池状态是否变化来决定是否继续

 *

 * @param onlyOne If true, interrupt at most one worker. This is

 * called only from tryTerminate when termination is otherwise

 * enabled but there are still other workers.  In this case, at

 * most one waiting worker is interrupted to propagate shutdown

 * signals in case(以免) all threads are currently waiting.

 * Interrupting any arbitrary thread ensures that newly arriving

 * workers since shutdown began will also eventually exit.

 * To guarantee eventual termination, it suffices to always

 * interrupt only one idle worker, but shutdown() interrupts all

 * idle workers so that redundant workers exit promptly, not

 * waiting for a straggler task to finish.

 *

 * onlyOne如果为true，最多interrupt一个worker

 * 只有当终止流程已经开始，但线程池还有worker线程时,tryTerminate()方法会做调用onlyOne为true的调用

 * （终止流程已经开始指的是：shutdown状态 且 workQueue为空，或者 stop状态）

 * 在这种情况下，最多有一个worker被中断，为了传播shutdown信号，以免所有的线程都在等待

 * 为保证线程池最终能终止，这个操作总是中断一个空闲worker

 * 而shutdown()中断所有空闲worker，来保证空闲线程及时退出

 */

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(); //解锁

    }

}

1

2

3

4

5

6

7

protected boolean tryAcquire(int unused) {

    if (compareAndSetState(0, 1)) {

        setExclusiveOwnerThread(Thread.currentThread());

        return true;

    }

    return false;

}

由此可以将worker划分为：
1、空闲worker：正在从workQueue阻塞队列中获取任务的worker
2、运行中worker：正在task.run()执行任务的worker

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

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

/**

 * Transitions to TERMINATED state if either (SHUTDOWN and pool

 * and queue empty) or (STOP and pool empty).  If otherwise

 * eligible to terminate but workerCount is nonzero, interrupts an

 * idle worker to ensure that shutdown signals propagate. This

 * method must be called following any action that might make

 * termination possible -- reducing worker count or removing tasks

 * from the queue during shutdown. The method is non-private to

 * allow access from ScheduledThreadPoolExecutor.

 *

 * 在以下情况将线程池变为TERMINATED终止状态

 * shutdown 且 正在运行的worker 和 workQueue队列 都empty

 * stop 且  没有正在运行的worker

 *

 * 这个方法必须在任何可能导致线程池终止的情况下被调用，如：

 * 减少worker数量

 * shutdown时从queue中移除任务

 *

 * 这个方法不是私有的，所以允许子类ScheduledThreadPoolExecutor调用

 */

final void tryTerminate() {

    //这个for循环主要是和进入关闭线程池操作的CAS判断结合使用的

    for (;;) {

        int c = ctl.get();

        /**

         * 线程池是否需要终止

         * 如果以下3中情况任一为true，return，不进行终止

         * 1、还在运行状态

         * 2、状态是TIDYING、或 TERMINATED，已经终止过了

         * 3、SHUTDOWN 且 workQueue不为空

         */

        if (isRunning(c) ||

            runStateAtLeast(c, TIDYING) ||

            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))

            return;

        /**

         * 只有shutdown状态 且 workQueue为空，或者 stop状态能执行到这一步

         * 如果此时线程池还有线程（正在运行任务，正在等待任务）

         * 中断唤醒一个正在等任务的空闲worker

         * 唤醒后再次判断线程池状态，会return null，进入processWorkerExit()流程

         */

        if (workerCountOf(c) != 0) { // Eligible to terminate 资格终止

            interruptIdleWorkers(ONLY_ONE); //中断workers集合中的空闲任务，参数为true，只中断一个

            return;

        }

        /**

         * 如果状态是SHUTDOWN，workQueue也为空了，正在运行的worker也没有了，开始terminated

         */

        final ReentrantLock mainLock = this.mainLock;

        mainLock.lock();

        try {

            //CAS：将线程池的ctl变成TIDYING（所有的任务被终止，workCount为0，为此状态时将会调用terminated()方法），期间ctl有变化就会失败，会再次for循环

            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {

                try {

                    terminated(); //需子类实现

                }

                finally {

                    ctl.set(ctlOf(TERMINATED, 0)); //将线程池的ctl变成TERMINATED

                    termination.signalAll(); //唤醒调用了 等待线程池终止的线程 awaitTermination()

                }

                return;

            }

        }

        finally {

            mainLock.unlock();

        }

        // else retry on failed CAS

        // 如果上面的CAS判断false，再次循环

    }

}

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

44

45

46

47

48

/**

 * Attempts to stop all actively executing tasks, halts the

 * processing of waiting tasks, and returns a list of the tasks

 * that were awaiting execution. These tasks are drained (removed)

 * from the task queue upon return from this method.

 * 尝试停止所有活动的正在执行的任务，停止等待任务的处理，并返回正在等待被执行的任务列表

 * 这个任务列表是从任务队列中排出（删除）的

 *

 * <p>This method does not wait for actively executing tasks to

 * terminate.  Use {@link #awaitTermination awaitTermination} to

 * do that.

 * 这个方法不用等到正在执行的任务结束，要等待线程池终止可使用awaitTermination()

 *

 * <p>There are no guarantees beyond best-effort attempts to stop

 * processing actively executing tasks.  This implementation

 * cancels tasks via {@link Thread#interrupt}, so any task that

 * fails to respond to interrupts may never terminate.

 * 除了尽力尝试停止运行中的任务，没有任何保证

 * 取消任务是通过Thread.interrupt()实现的，所以任何响应中断失败的任务可能永远不会结束

 *

 * @throws SecurityException {@inheritDoc}

 */

public List<Runnable> shutdownNow() {

    List<Runnable> tasks;

    final ReentrantLock mainLock = this.mainLock;

    mainLock.lock(); //上锁

    try {

        //判断调用者是否有权限shutdown线程池

        checkShutdownAccess();

        //CAS+循环设置线程池状态为stop

        advanceRunState(STOP);

        //中断所有线程，包括正在运行任务的

        interruptWorkers();

        tasks = drainQueue(); //将workQueue中的元素放入一个List并返回

    }

    finally {

        mainLock.unlock(); //解锁

    }

    //尝试终止线程池

    tryTerminate();

    return tasks; //返回workQueue中未执行的任务

}

1

2

3

4

5

6

7

8

9

10

private void interruptWorkers() {

    final ReentrantLock mainLock = this.mainLock;

    mainLock.lock();

    try {

        for (Worker w : workers)

            w.interruptIfStarted();

    } finally {

        mainLock.unlock();

    }

}

1

2

3

4

/**

 * Wait condition to support awaitTermination

 */

private final Condition termination = mainLock.newCondition();

1

2

3

4

5

6

7

8

9

executorService.shutdown();

try{

    while(!executorService.awaitTermination(500, TimeUnit.MILLISECONDS)) {

        LOGGER.debug("Waiting for terminate");

    }

}

catch (InterruptedException e) {

    //中断处理

}