五. ThreadPoolExecutor原理解析-关闭线程池

爱被打了一巴掌 2024-03-29 14:41 71阅读 0赞

### 前言 ###

关闭`ThreadPoolExecutor`的方法有`shutdown()`和`shutdownNow()`方法，本篇文章将对`ThreadPoolExecutor`的关闭进行分析。

### 正文 ###

#### 一. shutdown() ####

首先分析`shutdown()`方法，其实现如下。

public void shutdown() {
        
    	final ReentrantLock mainLock = this.mainLock;
    	mainLock.lock();
    	try {
        
    		checkShutdownAccess();
    		// 循环通过CAS方式将线程池状态置为SHUTDOWN
    		advanceRunState(SHUTDOWN);
    		// 中断空闲Worker
    		interruptIdleWorkers();
    		onShutdown();
    	} finally {
        
    		mainLock.unlock();
    	}
    	// 尝试终止线程池
    	tryTerminate();
    }

在`shutdown()`方法中首先会将线程池状态置为`SHUTDOWN`，然后调用`interruptIdleWorkers()`方法中断空闲`Worker`，最后调用`tryTerminate()`方法来尝试终止线程池。那么这里要解释一下什么是`空闲Worker`，先看一下`interruptIdleWorkers()`的实现。

private void interruptIdleWorkers() {
        
    	interruptIdleWorkers(false);
    }
    
    private void interruptIdleWorkers(boolean onlyOne) {
        
    	final ReentrantLock mainLock = this.mainLock;
    	mainLock.lock();
    	try {
        
    		for (Worker w : workers) {
        
    			Thread t = w.thread;
    			// 中断线程前需要先尝试获取Worker的锁
    			// 只能获取到空闲Worker的锁，所以shutdown()方法只会中断空闲Worker
    			if (!t.isInterrupted() && w.tryLock()) {
        
    				try {
        
    					t.interrupt();
    				} catch (SecurityException ignore) {
        
    				} finally {
        
    					w.unlock();
    				}
    			}
    			if (onlyOne)
    				break;
    		}
    	} finally {
        
    		mainLock.unlock();
    	}
    }

调用`interruptIdleWorkers()`方法中断`Worker`前首先需要尝试获取`Worker`的锁，已知`Worker`除了实现`Runnable`接口外，还继承于`AbstractQueuedSynchronizer`，因此`Worker`本身是一把锁，然后在`runWorker()`中`Worker`执行任务前都会先获取`Worker`的锁，这里看一下`Worker`的`lock()`方法的实现。

public void lock() {
         
    	acquire(1);
    }
    
    protected boolean tryAcquire(int unused) {
        
    	// 以CAS方式将state从0设置为1
    	if (compareAndSetState(0, 1)) {
        
    		setExclusiveOwnerThread(Thread.currentThread());
    		return true;
    	}
    	return false;
    }

可以发现，`Worker`在`lock()`中调用了`acquire()`方法，该方法由`AbstractQueuedSynchronizer`抽象类提供，在`acquire()`中会调用其子类实现的`tryAcquire()`方法，`tryAcquire()`方法会以**CAS**方式将**state**从0设置为1，因此这样的设计让`Worker`是一把不可重入锁。回到`interruptIdleWorkers()`方法，前面提到该方法中断`Worker`前会尝试获取`Worker`的锁，能够获取到锁才会中断`Worker`，而因为`Worker`是不可重入锁，所以正在执行任务的`Worker`是无法获取到锁的，只有那些没有执行任务的`Worker`的锁才能够被获取，因此所谓的中断`空闲Worker`，实际就是中断`没有执行任务的Worker`，那些执行任务的`Worker`在`shutdown()`方法被调用时不会被中断，这些`Worker`执行完任务后会继续从任务阻塞队列中获取任务来执行，直到任务阻塞队列为空，此时没有被中断过的`Worker`也会被删除掉，等到线程池中没有`Worker`以及任务阻塞队列没有任务后，线程池才会被终止掉。

对于`shutdown()`方法，一句话总结就是：**将线程池状态置为SHUTDOWN并拒绝接受新任务，等到线程池`Worker`数量为0，任务阻塞队列为空时，关闭线程池。**

#### 二. shutdownNow() ####

现在再来分析`shutdownNow()`方法。

public List<Runnable> shutdownNow() {
        
    	List<Runnable> tasks;
    	final ReentrantLock mainLock = this.mainLock;
    	mainLock.lock();
    	try {
        
    		checkShutdownAccess();
    		// 循环通过CAS方式将线程池状态置为STOP
    		advanceRunState(STOP);
    		// 中断所有Worker
    		interruptWorkers();
    		// 将任务阻塞队列中的任务获取出来并返回
    		tasks = drainQueue();
    	} finally {
        
    		mainLock.unlock();
    	}
    	// 尝试终止线程池
    	tryTerminate();
    	return tasks;
    }
    
    private void interruptWorkers() {
        
    	final ReentrantLock mainLock = this.mainLock;
    	mainLock.lock();
    	try {
        
    		// 中断线程池中所有Worker
    		for (Worker w : workers)
    			w.interruptIfStarted();
    	} finally {
        
    		mainLock.unlock();
    	}
    }

在`shutdownNow()`方法中首先会将线程池状态置为`STOP`，然后调用`interruptWorkers()`方法中断线程池中的所有`Worker`，接着调用`tryTerminate()`方法来尝试终止线程池，最后`shutdownNow()`方法会将任务阻塞队列中还未被执行的任务返回。`shutdownNow()`方法调用之后，线程池中的所有`Worker`都会被中断，包括正在执行任务的`Worker`，等到所有`Worker`都被删除之后，线程池即被终止，也就是说，`shutdownNow()`不会保证当前时刻正在执行的任务会被安全的执行完，并且会放弃执行任务阻塞队列中的所有任务。

#### 三. tryTerminate() ####

关于线程池的关闭，还有一个重要的方法，那就是前面多次提到的`tryTerminate()`方法，该方法能确保线程池可以被正确的关闭，其实现如下所示。

final void tryTerminate() {
        
    	for (;;) {
        
    		int c = ctl.get();
    		// 如果线程池状态为RUNNING，则没有资格终止线程池
    		// 如果线程池状态大于等于TIDYING，则没有资格终止线程池
    		// 如果线程池状态为SHUTDOWN但任务阻塞队列不为空，则没有资格终止线程池
    		if (isRunning(c) ||
    			runStateAtLeast(c, TIDYING) ||
    			(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
    			return;
    		// 线程池状态为SHUTDOWN且任务阻塞队列为空会执行到这里
    		// 线程池状态为STOP会执行到这里
    		// Worker数量不为0，表明当前还有正在执行任务的Worker或者空闲的Worker，此时中断一个空闲的Worker
    		// 在这里被中断的空闲Worker会在getTask()方法中返回null，从而执行processWorkerExit()，最终该Worker会被删除
    		// processWorkerExit()方法中又会调用tryTerminate()，因此将shutdown信号在空闲Worker之间进行了传播
    		if (workerCountOf(c) != 0) {
        
    			interruptIdleWorkers(ONLY_ONE);
    			return;
    		}
    
    		final ReentrantLock mainLock = this.mainLock;
    		mainLock.lock();
    		try {
        
    			// 将线程池状态置为TIDYING
    			if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
        
    				try {
        
    					// 终止线程池
    					terminated();
    				} finally {
        
    					// 将线程池状态最终置为TERMINATED
    					ctl.set(ctlOf(TERMINATED, 0));
    					termination.signalAll();
    				}
    				return;
    			}
    		} finally {
        
    			mainLock.unlock();
    		}
    	}
    }

在`tryTerminate()`方法的官方注释中给出了两种线程池会被终止的情况：

*  线程池的状态为**SHUTDOWN**，`Worker`数量为0，任务阻塞队列为空；
 *  线程池的状态为**STOP**，`Worker`数量为0。

官方注释中还说明在所有可能导致线程池终止的操作中都应该调用`tryTerminate()`方法来尝试终止线程池，因此**线程池中`Worker`被删除时**和**任务阻塞队列中任务被删除时**会调用`tryTerminate()`，以达到在线程池符合终止条件时及时终止线程池。

### 总结 ###

关闭`ThreadPoolExecutor`有两种方式，总结如下。

1.  `shutdown()`。调用`shutdown()`方法会首先将线程池状态置为**SHUTDOWN**并拒绝接受新任务，然后中断空闲`Worker`，等到线程池中`Worker`数量为0，任务阻塞队列为空时，线程池被真正关闭；
2.  `shutdownNow()`。调用`shutdownNow()`方法会首先将线程池状态置为**STOP**，然后中断所有`Worker`（包括正在执行任务的`Worker`），并将任务阻塞队列中还未被执行的任务返回，当线程池`Worker`数量为0时，线程池被真正关闭。

还有一点需要说明，`Worker`除了实现`Runnable`接口外，还继承于`AbstractQueuedSynchronizer`，因此`Worker`本身是一把锁，`Worker`执行任务前都会先获取`Worker`的锁，所以正在执行任务的`Worker`的锁是无法被获取的，换言之，只有没有执行任务的`Worker`的锁才能被获取，这些`Worker`就称为空闲`Worker`。