DelayedWorkQueue与ScheduledFutureTask

旧城等待， 2022-08-31 13:19 59阅读 0赞

> **关键词：**BlockingQueue    DelayedWorkQueue    ScheduledFutureTask    ScheduledThreadPoolExecutor

在**ScheduledThreadPoolExecutor**这种类型的**ThreadPoolExecutor**线程池中，使用到的阻塞队列为**DelayedWorkQueue**，并且需要了解**ScheduledFutureTask**。为方便ScheduledThreadPoolExecutor的学习，这里先介绍DelayedWorkQueue、ScheduledFutureTask。

--------------------

# **█ **DelayedWorkQueue #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

DelayedWorkQueue实现了**BlockingQueue**接口，其是在ScheduledThreadPoolExecutor中定义的一个内部类。

static class DelayedWorkQueue extends AbstractQueue<Runnable>
        implements BlockingQueue<Runnable>

从DelayedWorkQueue的类定义上面可以看出，泛型为Runnable，即DelayedWorkQueue队列中存放的元素类型为Runnable类型。

### 字段： ###

private static final int INITIAL_CAPACITY = 16;
    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();

*  **queue**

**RunnableScheduledFuture**类型的数组，从DelayedWorkQueue类的定义上已经看见，其声明的泛型类型为Runnable类型，实际上队列中存放的具体类型为RunnableScheduledFuture。

public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {
    
        boolean isPeriodic();
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2][]

*  **INITIAL\_CAPACITY**

数组queue的初始化容量大小，默认值为16。与其他的BlockingQueue实现不同，DelayedWorkQueue队列大小容量不是固定不变的，其是可以进行扩容的。而像**LinkedBlockingQueue**（请戳《[BlockingQueue之LinkedBlockingQueue][BlockingQueue_LinkedBlockingQueue]》）、**ArrayBlockingQueue**（请戳《[BlockingQueue之ArrayBlockingQueue][BlockingQueue_ArrayBlockingQueue]》）一但在创建的时候指定了队列的容量大小之后，队列的容量是无法改变的。

*  **size**

queue队列中实际存放的元素的个数。size的取值范围是0到queue数组的长度。

*  **leader**

Thread类型，定义正在等待从队列中取出队首元素的线程。

###  **方法：** ###

**（1）向队列中存放元素。**

*  **offer**

BlockingQueue约束在使用该方法向队列中添加元素，当队列满了无法继续添加元素时，元素添加失败，返回false。但DelayedWorkQueue的队列是具有扩容功能的，因此不会存在队列满了的情况，DelayedWorkQueue虽然实现了BlockingQueue接口，但对某些BlockingQueue中约束的行为进行了打破，没有遵循。

public boolean offer(Runnable x) {
        // 参数不允许为null，否则会抛出空指针异常。
        if (x == null)
            throw new NullPointerException();
        // 将Runnable类型的参数强转成RunnableScheduledFuture。    
        RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
        // 使用ReentrantLock独占锁保证线程安全
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 获取队列中已存放元素的个数
            int i = size;
            // 如果实际存放的元素已经达到了队列的容量，则进行扩容。
            if (i >= queue.length)
                grow();
            size = i + 1;
            // i==0，即size==0，队列中没有元素，此时新添加的元素就是队列的第一个元素。
            if (i == 0) {
                queue[0] = e;
                setIndex(e, 0);
            } else {
                // 将元素添加到队列的对应的下标问题
                siftUp(i, e);
            }
            if (queue[0] == e) {
                leader = null;
                available.signal();
            }
        } finally {
            lock.unlock();
        }
        return true;
    }

①RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;

将Runnable类型的参数强转成RunnableScheduledFuture，如果Runnable类型的参数x不是RunnableScheduledFuture类型的，会发生类型转换异常。可见虽然方法参数定义为Runnable类型，但使用时需要为RunnableScheduledFuture或其子类型。

②grow

当队列中存放的元素已经达到了队列数组的长度，对数组进行扩容。oldCapacity >> 1操作是对oldCapacity进行除以2，即扩容是对原先的数组容量扩容一半。如原先的数组容量为16，则扩容后数组容量为16+16/2 = 24。

private void grow() {
        int oldCapacity = queue.length;
        int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50%
        if (newCapacity < 0) // overflow
            newCapacity = Integer.MAX_VALUE;
        queue = Arrays.copyOf(queue, newCapacity);
    }

*  **put**

put方法直接调用offer方法。

public void put(Runnable e) {
        offer(e);
    }

*  **add**

add方法也是直接调用了offer方法。

public boolean add(Runnable e) {
        return offer(e);
    }

**（2）从队列中取出元素。**

*  **take**

从队列中取出队首元素，当队列中没有元素可以取出的时候或者虽然有元素可以取出但是没有达到延迟等待时长，方法会一直阻塞等待，直到有元素可以取出，返回该元素。

public RunnableScheduledFuture<?> take() throws InterruptedException {
        // 使用ReentrantLock独占锁保证线程安全
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (;;) {
                // 取出队列中的第一个元素，即队首元素
                RunnableScheduledFuture<?> first = queue[0];
                // 队首元素为null，说明队列中没有元素，让线程等待
                if (first == null)
                    available.await();
                else {
                    // 获取延迟的时间
                    long delay = first.getDelay(NANOSECONDS);
                    // delay<=0，即延迟等待时间已到，不需要再延迟等待了
                    if (delay <= 0)
                        return finishPoll(first);
                    first = null;
                    if (leader != null)
                        available.await();
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && queue[0] != null)
                available.signal();
            lock.unlock();
        }
    }

①long delay = first.getDelay(NANOSECONDS);

first.getDelay，即调用ScheduledFutureTask的getDelay方法，返回值的时间单位为纳秒。

②finishPoll

当延迟时间已到，不需要再延迟等待了，对该队首元素做响应的处理，然后返回。

private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
        int s = --size;
        RunnableScheduledFuture<?> x = queue[s];
        queue[s] = null;
        if (s != 0)
            siftDown(0, x);
        setIndex(f, -1);
        return f;
    }

③如果没有达到延迟时长，则需要继续延迟等待。

*  **poll**

从队列中取出队首元素，如果队列为空没有元素可以取出或者没有达到延迟等待时长，方法会直接返回null。

public RunnableScheduledFuture<?> poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 取出队首元素
            RunnableScheduledFuture<?> first = queue[0];
            if (first == null || first.getDelay(NANOSECONDS) > 0)
                return null;
            else
                return finishPoll(first);
        } finally {
            lock.unlock();
        }
    }

①first == null || first.getDelay(NANOSECONDS) > 0

first==null，即队首元素为null，队列中没有元素可以取出，此时方法直接返回null。first.getDelay(NANOSECONDS) > 0即还有剩余的延迟时长，还没有达到延迟等待时长，虽然此时有元素可以取出，但还是返回null。

②当队首有元素可以取出并且已经达到了延迟等待时长，将元素取出。

--------------------

# **█ **ScheduledFutureTask #

RunnableScheduledFuture的实现类，DelayedWorkQueue队列中存放的元素类型为RunnableScheduledFuture，但因为RunnableScheduledFuture是接口无法实例化对象，所以定义其子类ScheduledFutureTask用于创建RunnableScheduledFuture对象。所以DelayedWorkQueue中最终实际存放的类型为ScheduledFutureTask。ScheduledFutureTask也是在ScheduledThreadPoolExecutor中定义的一个内部类。

private class ScheduledFutureTask<V>
            extends FutureTask<V> implements RunnableScheduledFuture<V>{
        ......            
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 4][]

**time：**

全局变量time，设置经过多长时间该任务能够执行。

private long time;

在构造器中对time进行设置：

ScheduledFutureTask(Runnable r, V result, long ns) {
        ......
        this.time = ns;
        ......
    }

getDelay：

方法，对外暴露获取time值，以纳秒为单位。

public long getDelay(TimeUnit unit) {
        return unit.convert(time - now(), NANOSECONDS);
    }

run：

实现Runnable接口中定义的run方法。因为ScheduledFutureTask也间接实现了Runnable接口，因此在线程池中运行的时候会调用其run方法：

public void run() {
        boolean periodic = isPeriodic();
        if (!canRunInCurrentRunState(periodic))
            cancel(false);
        else if (!periodic)
            ScheduledFutureTask.super.run();
        else if (ScheduledFutureTask.super.runAndReset()) {
            setNextRunTime();
            reExecutePeriodic(outerTask);
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20220829/156ab98be1da40cf910ab392d2e2c457.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20220829/46dc0bec0d9848c3aaa5e2d40069ac64.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2]: /images/20220829/ee8ae604865943ccae2187d183d0d156.png
[BlockingQueue_LinkedBlockingQueue]: https://blog.csdn.net/weixin_50518271/article/details/118195710?spm=1001.2014.3001.5501
[BlockingQueue_ArrayBlockingQueue]: https://blog.csdn.net/weixin_50518271/article/details/118162245?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3]: /images/20220829/d211ac22b55b47de92ba3642a898ae85.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 4]: /images/20220829/511e8552cf5f4109a66cd15a1fb3e9ac.png