八. ScheduledThreadPoolExecutor原理解析-DelayedWorkQueue

墨蓝 2024-03-29 14:42 88阅读 0赞

### 前言 ###

`DelayedWorkQueue`是`ScheduledThreadPoolExecutor`线程池使用的任务阻塞队列。`DelayedWorkQueue`是基于小根堆实现的延时优先级队列，队列中的元素就是`ScheduledFutureTask`，因此`DelayedWorkQueue`的队列头节点任务总是最优先被执行的任务。本篇文章将对`DelayedWorkQueue`的实现原理进行分析。

### 正文 ###

先看一下`DelayedWorkQueue`的字段，如下所示。

static class DelayedWorkQueue extends AbstractQueue<Runnable>
    	implements BlockingQueue<Runnable> {
        
    
    	// 堆数组的初始大小
    	private static final int INITIAL_CAPACITY = 16;
    	
    	// 堆数组，数组中的元素实际上是ScheduledFutureTask
    	private RunnableScheduledFuture<?>[] queue =
    		new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
    	
    	private final ReentrantLock lock = new ReentrantLock();
    	
    	// 延时队列元素个数
    	private int size = 0;
    
    	// 在延时队列头等待任务的领导线程
    	private Thread leader = null;
    
    	private final Condition available = lock.newCondition();
    	
    	......
    	
    }

特别说明一下**leader**字段和**available**字段，首先是**leader**字段，表示在延时队列头等待任务的第一个线程，即如果延时队列头的任务需要被执行时，这个任务会被**leader**字段指向的线程获得。同时所有在延时队列头等待任务的线程，均会在**available**上进入等待状态，并且在延时队列头的任务需要被执行时或者延时队列头的任务被更新时唤醒所有在**available**上等待的线程。

已知`DelayedWorkQueue`是一个基于小根堆实现的延时优先级队列，那么往`DelayedWorkQueue`中插入和删除任务后，均需要保持堆的性质，在`DelayedWorkQueue`中，主要是`siftUp()`和`siftDown()`这两个方法来保持堆的性质，`siftUp()`是用于往`DelayedWorkQueue`中插入任务时来保持堆的性质，而`siftDown()`是用于`DelayedWorkQueue`弹出任务后保持堆的性质，其实现如下。

`siftUp()`实现如下所示。

private void siftUp(int k, RunnableScheduledFuture<?> key) {
        
    	while (k > 0) {
        
    		// 计算父节点索引
    		int parent = (k - 1) >>> 1;
    		RunnableScheduledFuture<?> e = queue[parent];
    		// 将插入元素与父节点元素进行比较
    		// 如果插入元素大于等于父节点元素，则循环结束
    		if (key.compareTo(e) >= 0)
    			break;
    		// 如果插入元素小于父节点元素，则插入元素与父节点元素互换位置
    		queue[k] = e;
    		setIndex(e, k);
    		k = parent;
    	}
    	queue[k] = key;
    	setIndex(key, k);
    }

`siftDown()`实现如下所示。

private void siftDown(int k, RunnableScheduledFuture<?> key) {
        
    	int half = size >>> 1;
    	while (k < half) {
        
    		// 计算左子节点索引，并将左子节点索引赋值给child
    		int child = (k << 1) + 1;
    		RunnableScheduledFuture<?> c = queue[child];
    		// 计算右子节点索引
    		int right = child + 1;
    		// 令c表示左子节点和右子节点中元素值更小的元素
    		// 令child表示左子节点和右子节点中元素值更小的节点索引
    		if (right < size && c.compareTo(queue[right]) > 0)
    			c = queue[child = right];
    		// 将当前元素值与c的值进行比较，如果当前元素值已经小于等于c的值，则退出循环
    		if (key.compareTo(c) <= 0)
    			break;
    		// 如果当前元素值大于c的值，则将当前元素与c互换位置
    		queue[k] = c;
    		setIndex(c, k);
    		k = child;
    	}
    	queue[k] = key;
    	setIndex(key, k);
    }

理解了`siftUp()`和`siftDown()`这两个方法之后，先来看一下`DelayedWorkQueue`中添加任务的实现。因为`DelayedWorkQueue`实现了`BlockingQueue`接口，因此对外提供了`put()`，`add()`，`offer()`和超时退出的`offer()`这四个方法来添加任务，但是因为`DelayedWorkQueue`在容量满时会进行扩容，可以当成一个无界队列来看待，所以`DelayedWorkQueue`的`put()`，`add()`和超时退出的`offer()`方法均是调用的`offer()`方法，下面看一下`offer()`方法的实现。

public boolean offer(Runnable x) {
        
    	if (x == null)
    		throw new NullPointerException();
    	RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
    	final ReentrantLock lock = this.lock;
    	lock.lock();
    	try {
        
    		int i = size;
    		// 延时队列已满时需要进行扩容处理
    		if (i >= queue.length)
    			// 扩容后容量为扩容前容量的1.5倍
    			grow();
    		size = i + 1;
    		if (i == 0) {
        
    			queue[0] = e;
    			setIndex(e, 0);
    		} else {
        
    			// 向延时队列插入任务
    			// 与向堆插入元素一样，将任务插入到任务堆的末尾节点，并逐步与父节点进行比较和交换，直到满足堆的性质为止
    			siftUp(i, e);
    		}
    		if (queue[0] == e) {
        
    			// 如果插入的任务最终成为延时队列头节点任务，那么重置领导线程leader并唤醒所有等待获取任务的线程
    			leader = null;
    			available.signal();
    		}
    	} finally {
        
    		lock.unlock();
    	}
    	return true;
    }

同样的，`DelayedWorkQueue`对外提供了`remove()`，`poll()`，`take()`和超时退出的`poll()`这四个方法来移除或获取任务，这里重点分析一下`take()`和超时退出的`poll()`这两个方法。

`take()`的实现如下所示。

public RunnableScheduledFuture<?> take() throws InterruptedException {
        
    	final ReentrantLock lock = this.lock;
    	lock.lockInterruptibly();
    	try {
        
    		for (;;) {
        
    			RunnableScheduledFuture<?> first = queue[0];
    			// 如果头节点任务为空，直接进入等待状态
    			if (first == null)
    				available.await();
    			else {
        
    				// delay表示延时队列头节点任务的剩余等待时间
    				long delay = first.getDelay(NANOSECONDS);
    				if (delay <= 0)
    					// 如果延时队列头节点任务的剩余等待时间小于等于0，则弹出头节点任务并保持堆性质
    					return finishPoll(first);
    				first = null;
    				if (leader != null)
    					// 如果已经存在领导线程，则进入等待状态
    					available.await();
    				else {
        
    					// 如果不存在领导线程，则将当前Worker的线程置为领导线程
    					Thread thisThread = Thread.currentThread();
    					leader = thisThread;
    					try {
        
    						// 等待delay的时间，即等到延时队列头任务可以执行
    						available.awaitNanos(delay);
    					} finally {
        
    						if (leader == thisThread)
    							leader = null;
    					}
    				}
    			}
    		}
    	} finally {
        
    		if (leader == null && queue[0] != null)
    			available.signal();
    		lock.unlock();
    	}
    }

`take()`方法中，首先判断延时队列头节点任务是否为空，如果为空则直接在**available**上进入等待状态，如果不为空则再判断任务是否已经可以执行，若可以执行则直接弹出任务并返回，若还不能执行那么就再判断领导线程是否已经存在，如果存在那么说明当前线程不是在延时队列头等待任务的第一个线程，需要在**available**上进入等待状态，如果不存在就说明当前线程是在延时队列头等待任务的第一个线程，需要将当前线程置为领导线程，然后在**available**上进入等待状态直到头节点任务可以执行。

超时退出的`poll()`和`take()`方法的大体实现一样，只是超时退出的`poll()`还需要额外加入对`Worker`从延时队列获取任务的等待时间的判断，其实现如下所示。

public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit)
    	throws InterruptedException {
        
    	// nanos表示Worker从延时队列获取任务的等待时间
    	long nanos = unit.toNanos(timeout);
    	final ReentrantLock lock = this.lock;
    	lock.lockInterruptibly();
    	try {
        
    		for (;;) {
        
    			RunnableScheduledFuture<?> first = queue[0];
    			// 延时队列为空时
    			// 如果nanos小于等于0，则直接返回null
    			// 如果nanos大于0，则进入等待状态并等待nanos的时间
    			if (first == null) {
        
    				if (nanos <= 0)
    					return null;
    				else
    					nanos = available.awaitNanos(nanos);
    			} else {
        
    				// delay表示延时队列头节点任务的剩余等待时间
    				long delay = first.getDelay(NANOSECONDS);
    				if (delay <= 0)
    					// 如果延时队列头节点任务的剩余等待时间小于等于0，则弹出头节点任务并保持堆性质
    					return finishPoll(first);
    				if (nanos <= 0)
    					// 如果Worker从延时队列获取任务的等待时间小于等于0，则返回null
    					return null;
    				first = null;
    				// 如下情况会进入等待状态并等待nanos的时间
    				// Worker从延时队列获取任务的等待时间小于延时队列头节点任务的剩余等待时间
    				// Worker从延时队列获取任务的等待时间大于等于延时队列头节点任务的剩余等待时间，但领导线程不为空
    				if (nanos < delay || leader != null)
    					nanos = available.awaitNanos(nanos);
    				else {
        
    					// 如果Worker从延时队列获取任务的等待时间大于等于延时队列头节点任务的剩余等待时间，且领导线程为空
    					// 将领导线程置为当前Worker的线程
    					Thread thisThread = Thread.currentThread();
    					leader = thisThread;
    					try {
        
    						// 等待delay的时间，即等到延时队列头任务可以执行
    						long timeLeft = available.awaitNanos(delay);
    						// 重新计算Worker从延时队列获取任务的等待时间
    						nanos -= delay - timeLeft;
    					} finally {
        
    						if (leader == thisThread)
    							leader = null;
    					}
    				}
    			}
    		}
    	} finally {
        
    		if (leader == null && queue[0] != null)
    			available.signal();
    		lock.unlock();
    	}
    }

### 总结 ###

关于`DelayedWorkQueue`，总结如下。

1.  `DelayedWorkQueue`是一个基于`小根堆`实现的延时优先级队列；
2.  `DelayedWorkQueue`的堆数组初始大小为16。