Java线程池的实现——底层源码原理

这里以常用的实现类ThreadPoolExecutor来讲述原理。

构造方法

共有四个构造方法，最终都是调用七个核心参数的构造方法，如果一些参数省略，比如拒绝策略，创造线程工厂，那么会使用默认的拒绝策略（抛出异常并且线程池停止运行），默认的创造线程工厂。

线程池状态值

通过移位获得值，效率高。
RUNNING等于-1<<29
SHUWDOWN等于0<<29
STOP等于1<<29
TYDING等于2<<29
TERMINATED等于3<<29

ctl，同时表示线程池状态和线程数

ctl是一个AtomicInteger，保证线程安全。
本质上，通过高3位表示线程池状态，低29位表示线程数，因此能够通过位运算很快得到线程状态和线程数。
ctl初始值是RUNNING和0进行位或，结果是RUNNING。
ctlOf，是将线程状态和线程数进行位或
runStateOf，计算线程池状态，输入参数为ctl.get()，计算过程为：ctl.get() & ~ capacity
workerCountOf，计算线程数，输入为ctl.get()，计算过程为：ctl.get() & capacity
~capacity的前三位是1，后面全是0

为什么要使用一个ctl表示两个变量？
共享变量在多线程条件下存在线程安全问题，ctl是采用CAS+volatile实现的，写少读多，且只能保证一个变量的原子性。

execute方法

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        if (isRunning(c) && workQueue.offer(command)) {
            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);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

相当于线程池执行原理：
1、判断输入参数是否为空，为空抛出异常
2、判断线程数是否小于核心线程数，小于则创建核心线程，并将任务交给这个核心线程，成功后返回；如果没有成功，比如创建失败，则继续下面逻辑
3、判断线程池状态为RUNNING并且将任务添加到任务队列是否成功，如果线程池不是RUNNING则把任务从队列移除，如果线程数为0则创建新线程
4、继续创建线程，如果创建失败（达到最大线程数），则使用拒绝策略；没有失败则直接返回

addWorker

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
        for (;;) {
            int wc = workerCountOf(c);
            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;
        }
    }
    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 {
                // Recheck while holding lock.
                // Back out on ThreadFactory failure or if
                // shut down before lock acquired.
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}
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

Worker类

Worker实现了Runnable接口，有两个成员属性，一个是Thread，一个是传入任务，run方法是调用runWorker方法。

runWorker方法

runWorker是ThreadPoolExecutor类的成员方法。
核心是task.run()。
具有切面编程的特点，有beforeExecute和afterExecute方法，本身是空实现，可以继承ThreadPoolExecutor并重写这两个方法进行增强。

getTask方法

workQueue.take一直等待获取任务，workQueue.poll超时等待获取任务。

线程池如何复用线程？

在runWorker方法中，通过while循环实现，当前任务不为空则执行，如果为空则不停在任务队列中获取任务并执行。

while (task != null || (task = getTask()) != null) {
	...
	task.run();执行任务
	...
}
1
2
3
4
5

线程池大小怎么变化？

默认情况下，核心线程数不会超时回收，最大线程数会被超时回收，但是如果设置参数allowCoreThreadTimeOut为true，核心线程也可以被超时回收。
在runWorker方法中，线程会执行任务，并不断在任务队列中取任务，但是如果任务队列为空时，while循环会跳出，执行processWorkerExit方法，在HashSet中移除线程。