ThreadPoolExecutor学习笔记（二）

发表于

书接上回，在addWorker 方法中，添加成功后会启动 Worker 中的线程

1
2
3
4
5
6
7
ini复制代码w = new Worker(firstTask);
final Thread t = w.thread;
...
if (workerAdded) {
    t.start();//启动worker中的线程
    workerStarted = true;
}

接下来看看run方法做了啥

1
2
3
csharp复制代码public void run() {
    runWorker(this);
}

runWorker

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
ini复制代码final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    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 ((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);//completedAbruptly 如果是用户业务导致的异常，为true
    }
}

启动线程的方法还是挺简单的。

processWorkerExit

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
scss复制代码private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
        decrementWorkerCount();//工作线程数减一

    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);//从workers中移除
    } finally {
        mainLock.unlock();
    }

    tryTerminate();

    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            //处理worker退出的时候，如果任务队列还有任务，保证至少有一个线程在执行任务
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && ! workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return; // replacement not needed
        }
        addWorker(null, false);//如果是用户线程导致的异常，补充一个worker
    }
}

shutdown

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
scss复制代码public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        //CAS改变线程池状态为SHUTDOWN
        advanceRunState(SHUTDOWN);
        //中断所有空闲线程
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

interruptIdleWorkers中断空闲线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
java复制代码private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            //w.tryLock()能拿到锁说明是空闲线程，在runWorker里面有 w.lock(); 和w.unlock();包裹的运行任务的代码，空闲线程阻塞在runWorker的task = getTask()这


            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } 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
scss复制代码final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
        //当前线程池状态判断
        if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
        //当前worker大于零，停止一个空闲线程，保证有一个线程能执行接下来的线程池的状态转换的代码
        if (workerCountOf(c) != 0) { // Eligible to terminate
            interruptIdleWorkers(ONLY_ONE);
            return;
        }
        // 执行线程池的状态转换
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            //把线程池状态转换为TIDYING状态
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    //执行terminated钩子函数
                    terminated();
                } finally {
                    //把线程池状态转换为 TERMINATED 状态
                    ctl.set(ctlOf(TERMINATED, 0));
                    termination.signalAll();
                }
                return;
            }
        } finally {
            mainLock.unlock();
        }
        // else retry on failed CAS
    }
}

shutdownNow

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
scss复制代码public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        //CAS改变线程池状态为 STOP
        advanceRunState(STOP);
        //中断所有线程
        interruptWorkers();
        //获取所有未执行的任务
        tasks = drainQueue();
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    //返回所有未执行的任务
    return tasks;
}

shutdownNow关闭线程池流程：

  1. 把线程池状态改为 STOP ，相当于把所有从 runWorker 进来的口子给拦住了，因为在runWorker中有如下判断：
1
2
3
4
5
less复制代码if ((runStateAtLeast(ctl.get(), STOP) ||
     (Thread.interrupted() &&
      runStateAtLeast(ctl.get(), STOP))) &&
    !wt.isInterrupted())
    wt.interrupt();

如果有线程在关闭线程池的时候，还在task.run();中执行业务代码，当线程被中断后，再次获取任务的时候会抛出异常，举个例子：
被中断的线程再次运行到queue.take();的时候直接抛出异常

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
scss复制代码public static void main(String[] args) {
    BlockingQueue queue = new ArrayBlockingQueue(1);
    Thread thread = new Thread(() -> {
        long count = 0;
        for (;;){
            if (count < 160) {
                count ++;
                System.out.println(count);
            }
            break;
        }
        try {
            queue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    thread.start();
    thread.interrupt();
    System.out.println(thread.isInterrupted());

}
返回结果：
true
1
java.lang.InterruptedException
	at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1220)
	at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:335)
	at java.util.concurrent.ArrayBlockingQueue.take(ArrayBlockingQueue.java:400)
	at com.networkbench.tingyun.demo.thread.Main.lambda$main$0(Main.java:19)
	at com.networkbench.tingyun.demo.thread.Main$$Lambda$1/1936628443.run(Unknown Source)
	at java.lang.Thread.run(Thread.java:745)
  1. 中断所有已经启动的现场
1
scss复制代码interruptWorkers();
  1. 获取所有未执行的任务
1
ini复制代码tasks = drainQueue();
  1. 更改线程池状态
1
scss复制代码tryTerminate()

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%