四. ThreadPoolExecutor原理解析-执行任务

约定不等于承诺〃 2024-03-30 17:41 81阅读 0赞

### 前言 ###

`ThreadPoolExecutor`执行任务的入口方法是`execute()`，本篇文章将结合`ThreadPoolExecutor`部分源码，对`ThreadPoolExecutor`执行任务的流程进行分析。

### 正文 ###

`ThreadPoolExecutor`中执行任务的入口方法为`execute()`，其实现如下。

public void execute(Runnable command) {
        
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        // 如果Worker数量小于核心线程数，则创建Worker并执行任务
        if (workerCountOf(c) < corePoolSize) {
        
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        // 如果Worker数量大于等于核心线程数，则将任务添加到任务阻塞队列
        if (isRunning(c) && workQueue.offer(command)) {
        
            int recheck = ctl.get();
            // 如果线程池状态突然不再是RUNNING，则尝试将任务从任务阻塞队列中删除，删除成功则为该任务执行拒绝策略
            if (! isRunning(recheck) && remove(command))
                reject(command);
            // 如果线程池中Worker数量突然为0，则创建一个Worker来执行任务
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        // 执行到这里表示线程池状态已经不是RUNNING或者任务阻塞队列已满
        // 此时尝试新建一个Worker来执行任务
        // 如果新建一个Worker来执行任务失败，表明线程池状态不再是RUNNING或者Worker数量已经达到最大线程数，此时执行拒绝策略
        else if (!addWorker(command, false))
            reject(command);
    }

在`execute()`中会根据`Worker`数量和线程池状态来决定是新建`Worker`来执行任务还是将任务添加到任务阻塞队列。新建`Worker`来执行任务的实现如下所示。

private boolean addWorker(Runnable firstTask, boolean core) {
        
        // 标记外层for循环
        retry:
        for (;;) {
        
            int c = ctl.get();
            // 获取线程池状态
            int rs = runStateOf(c);
    
            // 线程池状态为RUNNING时，可以创建Worker
            // 线程池状态为SHUTDOWN，且任务阻塞队列不为空时，可以创建初始任务为null的Worker
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
    
            for (;;) {
        
                // 获取Worker数量
                int wc = workerCountOf(c);
                
                // 如果Worker数量大于CAPACITY，拒绝创建Worker
                // core为true表示创建核心线程Worker，如果Worker数量此时已经大于等于核心线程数，则拒绝创建Worker，转而应该将任务添加到任务阻塞队列
                // core为false表示创建非核心线程Worker，如果Worker数量此时已经大于等于最大线程数，则拒绝创建Worker，转而应该执行拒绝策略
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // 以CAS方式将Worker数量加1
                // 加1成功表明无竞争发生，从外层for循环跳出
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                // 加1失败表明有竞争发生，此时需要重新获取ctl的值
                c = ctl.get();
                // 重新获取ctl后如果发现线程池状态发生了改变，此时重新执行外层for循环，即需要基于新的线程池状态判断是否允许创建Worker
                // 重新获取ctl后如果线程池状态未发生改变，则继续执行内层for循环，即尝试再一次以CAS方式将Worker数量加1
                if (runStateOf(c) != rs)
                    continue retry;
            }
        }
    
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
        
            // 创建一个Worker
            w = new Worker(firstTask);
            // 获取Worker的线程
            final Thread t = w.thread;
            if (t != null) {
        
                final ReentrantLock mainLock = this.mainLock;
                // 由于线程池中存储Worker的集合为HashSet，因此将Worker添加到Worker集合时需要获取全局锁保证线程安全
                mainLock.lock();
                try {
        
                    // 再一次获取线程池状态
                    int rs = runStateOf(ctl.get());
    
                    // 如果线程池状态还是为RUNNING或者线程池状态为SHUTDOWN但创建的Worker的初始任务为null，则允许将创建出来的Worker添加到集合
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
        
                        // 检查一下Worker的线程是否可以启动（处于活动状态的线程无法再启动）
                        if (t.isAlive())
                            throw new IllegalThreadStateException();
                        // 将Worker添加到Worker集合
                        workers.add(w);
                        int s = workers.size();
                        // largestPoolSize用于记录线程池最多存在过的Worker数
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
        
                    mainLock.unlock();
                }
                if (workerAdded) {
        
                    // 启动Worker线程
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
        
            if (! workerStarted)
                // Worker线程没有成功启动起来，此时需要对该Worker的创建执行回滚操作
                addWorkerFailed(w);
        }
        return workerStarted;
    }

`addWorker()`方法中只允许两种情况可以创建`Worker`。

*  线程池状态为**RUNNING**，可以创建`Worker`；
 *  线程池状态为**SHUTDOWN**，且任务阻塞队列不为空，可以创建初始任务为**null**的`Worker`。

一旦`Worker`创建成功，就会将`Worker`的线程启动，如果`Worker`创建失败或者`Worker`的线程启动失败，则会调用`addWorkerFailed()`方法执行回滚操作，其实现如下所示。

private void addWorkerFailed(Worker w) {
        
    	final ReentrantLock mainLock = this.mainLock;
    	mainLock.lock();
    	try {
        
    		// 如果Worker添加到了Worker集合中，则将Worker从Worker集合中删除
    		if (w != null)
    			workers.remove(w);
    		// 以CAS方式将Worker数量减1
    		decrementWorkerCount();
    		// 尝试终止线程池
    		tryTerminate();
    	} finally {
        
    		mainLock.unlock();
    	}
    }

由于`Worker`自身实现了`Runnable`，因此`Worker`自身就是一个任务，实际上`Worker`的线程执行的任务就是`Worker`本身，因此`addWorker()`中将`Worker`的线程启动时，会调用`Worker`的`run()`方法，其实现如下。

public void run() {
        
        runWorker(this);
    }

在`Worker`的`run()`方法中调用了`ThreadPoolExecutor`的`runWorker()`方法，其实现如下所示。

final void runWorker(Worker w) {
        
    	Thread wt = Thread.currentThread();
    	Runnable task = w.firstTask;
    	w.firstTask = null;
    	w.unlock();
    	boolean completedAbruptly = true;
    	try {
        
    		// 如果task为null，则从任务阻塞队列中获取任务
    		// 通常Worker启动时会先执行初始任务，然后再去任务阻塞队列中获取任务
    		while (task != null || (task = getTask()) != null) {
        
    			w.lock();
    			// 线程池正在停止时，需要确保当前Worker的线程是被中断的
    			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 {
        
    		// Worker执行任务发生异常或者从getTask()中获取任务为空时会执行这里的逻辑
    		// processWorkerExit()会将Worker从Worker集合中删除，并尝试终止线程池
    		processWorkerExit(w, completedAbruptly);
    	}
    }

`runWorker()`方法就是先让`Worker`将初始任务（如果有的话）执行完，然后循环从任务阻塞队列中获取任务来执行，如果`Worker`执行任务发生异常或者从任务阻塞队列获取任务失败（获取到的任务为**null**），则调用`processWorkerExit()`方法来将自身从`Worker`集合中删除。下面先看一下`getTask()`方法的实现。

private Runnable getTask() {
        
    	boolean timedOut = false;
    
    	for (;;) {
        
    		int c = ctl.get();
    		int rs = runStateOf(c);
    
    		// 如果线程池状态为SHUTDOWN，且任务阻塞队列为空，则不再允许从任务阻塞队列中获取任务
    		// 如果线程池状态为STOP，则不再允许从任务阻塞队列中获取任务
    		if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
        
    			decrementWorkerCount();
    			return null;
    		}
    
    		int wc = workerCountOf(c);
    
    		// 如果allowCoreThreadTimeOut为true，或者当前线程数大于核心线程数，此时timed为true，表明从任务阻塞队列以超时退出的方式获取任务
    		boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
    		// 如果当前线程数大于最大线程数，则当前Worker应该被删除
    		// 如果当前Worker上一次从任务阻塞队列中获取任务时超时退出，且任务阻塞队列现在还是为空，则当前Worker应该被删除
    		if ((wc > maximumPoolSize || (timed && timedOut))
    			&& (wc > 1 || workQueue.isEmpty())) {
        
    			if (compareAndDecrementWorkerCount(c))
    				return null;
    			continue;
    		}
    
    		try {
        
    			// 从任务阻塞队列中获取任务
    			Runnable r = timed ?
    				workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
    				workQueue.take();
    			if (r != null)
    				// 获取到任务则返回该任务
    				return r;
    			// timedOut为true表明Worker上一次从任务阻塞队列中获取任务时超时退出
    			timedOut = true;
    		} catch (InterruptedException retry) {
        
    			timedOut = false;
    		}
    	}
    }

`getTask()`方法在如下情况不允许`Worker`从任务阻塞队列中获取任务。

*  线程池状态为**SHUTDOWN**，且任务阻塞队列为空；
 *  线程池状态为**STOP**。

如果`Worker`有资格从任务阻塞队列获取任务，那么当`allowCoreThreadTimeOut`为**true**，或者当前线程数大于核心线程数时，`Worker`以超时退出的方式获取任务，否则`Worker`以一直阻塞的方式获取任务。

当`Worker`在`getTask()`方法中获取任务失败时，`getTask()`方法会返回**null**，从而导致`Worker`会执行`processWorkerExit()`方法来删除自身，其实现如下所示。

private void processWorkerExit(Worker w, boolean completedAbruptly) {
        
    	// completedAbruptly为true表明是执行任务时发生异常导致Worker需要被删除
    	if (completedAbruptly)
    		// 修正Worker数量
    		decrementWorkerCount();
    
    	final ReentrantLock mainLock = this.mainLock;
    	mainLock.lock();
    	try {
        
    		completedTaskCount += w.completedTasks;
    		// 将Worker从Worker集合中删除
    		workers.remove(w);
    	} finally {
        
    		mainLock.unlock();
    	}
    
    	// 尝试终止线程池
    	tryTerminate();
    
    	int c = ctl.get();
    	if (runStateLessThan(c, STOP)) {
        
    		if (!completedAbruptly) {
        
    			int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
    			if (min == 0 && ! workQueue.isEmpty())
    				min = 1;
    			if (workerCountOf(c) >= min)
    				return;
    		}
    		addWorker(null, false);
    	}
    }

`Worker`在`processWorkerExit()`方法中删除自身之后，还会调用`tryTerminate()`尝试终止线程池，`tryTerminate()`方法很精髓，后面会对其进行详细分析，这里暂且不谈。至此，`Worker`的创建，执行任务，获取任务和删除的整个流程已经大体分析完毕。

### 总结 ###

`ThreadPoolExecutor`执行任务，第一步是根据`Worker`数量来决定是新建`Worker`来执行任务还是将任务添加到任务阻塞队列，这里的判断规则如下。

1.  如果`Worker`数量小于核心线程数，则创建`Worker`来执行任务；
2.  如果`Worker`数量大于等于核心线程数，则将任务添加到任务阻塞队列；
3.  如果任务阻塞队列已满，则创建`Worker`来执行任务；
4.  如果`Worker`数量已经达到最大线程数，此时执行任务拒绝策略。

当要新建`Worker`来执行任务时，只有两种情况可以新建`Worker`，如下所示。

1.  线程池状态为**RUNNING**，可以创建`Worker`；
2.  线程池状态为**SHUTDOWN**，且任务阻塞队列不为空，可以创建初始任务为**null**的`Worker`。

`Worker`自身实现了`Runnable`，且`Worker`持有一个线程，当`Worker`启动时，就是启动`Worker`持有的线程，而这个线程执行的任务就是`Worker`自身。

`Worker`启动后，会首先执行自己的初始任务，然后再去任务阻塞队列中获取任务。