ThreadPoolExecutor源码分析之execute方法_threadpoolexecutor用execute

几种状态二进制：

RUNNING:
11100000000000000000000000000000
SHUTDOWN:
00000000000000000000000000000000
STOP:
00100000000000000000000000000000
TIDYING:
01000000000000000000000000000000
TERMINATED:
01100000000000000000000000000000
他们的高3位分别为：
111
000
010
100
110

用来做二进制操作的变量二进制：
CAPACITY： 00011111111111111111111111111111

ctl变量是一个AtomicInteger类型，高3位存储线程池状态，低29位存储线程数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

知道了这几个二进制变量，然后具体分析一下有关位运算的这几个方法：
private static int runStateOf(int c) { return c & ~CAPACITY; }
runStateOf用来获取高3位的值，即线程池状态
private static int workerCountOf(int c) { return c & CAPACITY; }
workerCountOf用来获取低29位的值，即工作线程数量
private static int ctlOf(int rs, int wc) { return rs | wc; }
ctlOf方法用来组合线程状态和线程数量为一个值，rs为线程状态值，wc为线程数量

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;
		// 添加失败，因为并发导致工作线程已经大于等于核心线程数，然后重新获取ctl值
		c = ctl.get();
	}
	// 判断线程池状态是否是RUNNING，并向阻塞队列中添加任务
	if (isRunning(c) && workQueue.offer(command)) {
		// 因为并发的原因，而且C变量没有锁，实际状态可能已经发生了变更，重新获取线程池状态检查
		int recheck = ctl.get();
		// 如果线程池不处于RUNNING状态，然后从阻塞队列移除当前任务
		if (! isRunning(recheck) && remove(command))
			reject(command);
		// TODO 工作线程等于0，则尝试添加工作线程处理阻塞队列中的任务
		else if (workerCountOf(recheck) == 0)
			addWorker(null, false);
	}
	// 这个 else if 主要处理阻塞队列满了，向线程池添加非核心线程处理任务。情形一：线程池不为RUNNING，情形一：线程池为RUNNING但是阻塞队列已经满了
	// 添加一个非核心工作线程并且立即处理当前任务
	// 情形一肯定添加失败，addWorker会通过cas判断当前线程池状态，如果不为RUNNING会返回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
20
21
22
23
24
25
26
27
28
29
30

代码中的注释非常详细, 这里再简要概括一下. execute()方法主要分为以下四种情况:
情况1: 如果线程池内的有效线程数少于核心线程数 corePoolSize, 那么就创建并启动一个线程来执行新提交的任务.
情况2: 如果线程池内的有效线程数达到了核心线程数 corePoolSize, 并且线程池内的阻塞队列未满, 那么就将新提交的任务加入到该阻塞队列中.
情况3: 如果线程池内的有效线程数达到了核心线程数 corePoolSize 但却小于最大线程数 maximumPoolSize, 并且线程池内的阻塞队列已满, 那么就创建并启动一个线程来执行新提交的任务.
情况4: 如果线程池内的有效线程数达到了最大线程数 maximumPoolSize, 并且线程池内的阻塞队列已满, 那么就让 RejectedExecutionHandler 根据它的拒绝策略来处理该任务, 默认的处理方式是直接抛异常.

要想彻底看懂 execute()方法的逻辑, 就必须要先大致了解 addWorker()方法的逻辑以及该方法分别在上述三组赋值情况下各自到底做了什么事情. 下面我们还是来分析一下 addWorker()方法的源码
// addWorker的任务是：添加工作线程，分3种情况
// 1.firstTask有值，core为true
// 2.firstTask有值，core为false
// 3.firstTask为null，core为false

private boolean addWorker(Runnable firstTask, boolean core) {
	retry:
	for (;;) {
		int c = ctl.get();
		int rs = runStateOf(c);

		// 线程处于以下几种情况时，才继续往下走，否则直接返回false
		// 1. 处于 RUNNING 状态
		// 2. 处于SHUTDOWN 状态 并且 firstTask == null 并且 workQueue不为空
		// 第2种情况，是因为在上面的execute方法中判断了工作线程为0，然后调用了addWorker方法，处理阻塞队列中的没有处理完的任务，所有需要继续往下走来添加工作线程
		if (rs >= SHUTDOWN &&
			! (rs == SHUTDOWN &&
			   firstTask == null &&
			   ! workQueue.isEmpty()))
			return false;
		// 如果线程池内的有效线程数大于或等于了理论上的最大容量 CAPACITY 或者实际
		// 设定的最大容量, 就返回 false直接结束该方法. 这样同样没有创建新线程, 
		// 新提交的任务也同样未被执行.
		// (core ? corePoolSize : maximumPoolSize) 表示如果 core为 true,
		// 那么实际设定的最大容量为 corePoolSize, 反之则为 maximumPoolSize.
		for (;;) {
			int wc = workerCountOf(c);
			if (wc >= CAPACITY ||
				wc >= (core ? corePoolSize : maximumPoolSize))
				return false;
			// 通过cas操作，修改工作线程数量+1，修改成功后，跳出retry循环，继续执行retry循环后面的代码；
			// 如果因为并发导致c变量的高3位状态值或者低29位的线程数量值改变了，则继续往下走，重新获取c变量的值
			if (compareAndIncrementWorkerCount(c))
				break retry;
			c = ctl.get();
			// 判断高3位的状态值是否改变，如果改变，跳会retry循环；
			// 否则就是低29位的线程数量值改变了，继续执行当前循环
			if (runStateOf(c) != rs)
				continue retry;
		}
	}

	// 走到这里说明线程状态可能有两种情况：1. RUNNING 状态  2. SHUTDOWN 状态 并且 firstTask == null 并且 workQueue不为空
	boolean workerStarted = false;
	boolean workerAdded = false;
	Worker w = null;
	try {
		// 创建Worker对象，并把第一个任务firstTask传给他
		w = new Worker(firstTask);
		final Thread t = w.thread;
		if (t != null) {
			// 线程池自己的一个可重入锁，锁住下面代码，防止并发
			final ReentrantLock mainLock = this.mainLock;
			mainLock.lock();
			try {
				// 获取线程状态值
				int rs = runStateOf(ctl.get());
				// 如果线程状态是RUNNING 或者 SHUTDOWN且firstTask == null
				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. 由于 t指向 w.thread所引用的对象, 所以相当于启动的是 w.thread所引用的线程对象.
                // 而 w是 Runnable 的实现类, w.thread 是以 w作为 Runnable参数所创建的一个线程对象, 所以启动
                // w.thread所引用的线程对象, 也就是要执行 w 的 run()方法.        
				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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80