DelayQueue源码简析

阳光穿透心脏的1/2处 2023-02-21 04:51 15阅读 0赞

DelayQueue延迟队列，意思是加入队列的元素可以设置一个延迟时间，当延迟时间到了才可以从队列中出队这个元素，先写一个demo来看看要如何使用DelayQueue

public class DelayQueueTest { 
        static class MyDelayed implements Delayed { 
            private final long delayTime ; //延迟时间
            private final long expire;     //到期时间
            private String taskName ;      //任务名称
    
            public MyDelayed(long delayTime, String taskName) { 
                this.delayTime = delayTime;
                this.expire = System.currentTimeMillis() + delayTime;
                this.taskName = taskName;
            }
    
            @Override
            public long getDelay(TimeUnit unit) { 
                return unit.convert(expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
            }
    
            @Override
            public int compareTo(Delayed o) { 
                return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
            }
    
            @Override
            public String toString() { 
                return "MyDelayed{" +
                        "delayTime=" + delayTime +
                        ", expire=" + expire +
                        ", taskName='" + taskName + '\'' +
                        '}';
            }
        }
    
        public static void main(String[] args) throws Exception { 
            MyDelayed delayed1 = new MyDelayed(1000l,"task1");
            MyDelayed delayed2 = new MyDelayed(4000l,"task2");
            MyDelayed delayed3 = new MyDelayed(2000l,"task3");
    
            DelayQueue<MyDelayed> queue = new DelayQueue<>();
            queue.put(delayed1);
            queue.put(delayed2);
            queue.put(delayed3);
    
            for (Object o : queue) { 
                Delayed delayed = queue.take();
                System.out.println(delayed.toString());
            }
        }
    }

从demo中可以看到，加入队列的元素，都实现了Delayed接口，同时需要实现getDelay方法和compareTo方法，从DelayQueue的类定义中我们可以找到这么做的原因

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
        implements BlockingQueue<E>

可以看到DelayQueue中的泛型实现了Delayed接口，里面提供了一个需要实现的getDelay方法，这个方法是为了返回剩余的延迟时间，同时  
Delayed接口继承了Comparable接口，至于为什么要实现这个接口呢？是因为DelayQueue在入队的时候，其实使用的是PriorityQueue优先队列来实现的，所以入队时需要进行一个排序，这里会用到Comparable接口

demo中给DelayQueue中的元素定义了不同的延迟时间，出队时是按照延迟时间的长短来出队的（也就是你定义的compare方法的逻辑），延迟时间最短的先出队，所以demo最后的输出为

MyDelayed{ delayTime=1000, expire=1593332294802, taskName='task1'}
    MyDelayed{ delayTime=2000, expire=1593332295802, taskName='task3'}
    MyDelayed{ delayTime=4000, expire=1593332297802, taskName='task2'}

了解了DelayQueue的基本用法之后，我们再来分析一下他的源码

public void put(E e) { 
       offer(e);
    }
    
    public boolean offer(E e) { 
    	// 这个lock是类中定义的
        final ReentrantLock lock = this.lock;
        lock.lock();
        try { 
        	// 往优先队列中加入元素
            q.offer(e);
           	// 如果添加的元素是队列里队首元素的话
            if (q.peek() == e) { 
            	// leader是一个Thread，将leader线程置为null
                leader = null;
                // available是lock锁的一个Condition
                available.signal();
            }
            return true;
        } finally { 
            lock.unlock();
        }
    }

关于优先队列出入队的具体步骤在我的博客 [PriorityBlockingQueue源码简析][PriorityBlockingQueue] 中都有具体的描述，感兴趣的同学可以看一下

关于leader和available我们现在还不清楚到底有什么作用，所以我们带着问题接着往下看出队的源码

public E take() throws InterruptedException { 
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try { 
            for (;;) { 
            	// 获取队列的头元素
                E first = q.peek();
                // 如果队列头元素是null的话，说明队列中已经没有元素了，那么available则调用await方法阻塞线程，阻止线程继续出队
                // 而在入队的逻辑中，判断队列中已经有元素的话，就唤醒被阻塞的出队线程继续执行出队操作
                if (first == null)
                    available.await();
                else { 
                	// 获取元素的延迟时间
                    long delay = first.getDelay(NANOSECONDS);
                    // 如果延迟时间小于等于0（取决于getDelay方法的实现），那么就从优先队列中将元素出队
                    if (delay <= 0)
                        return q.poll();
                    // 将first的引用清空，在后面的等待过程中不再保留这个引用，方便垃圾回收
                    first = null; // don't retain ref while waiting
                    // 如果leader线程不为空的话，则调用available阻塞线程
                    if (leader != null)
                        available.await();
                    else { 
                    	// 进入else判断的话，说明leader线程为null，这里将当前线程赋值给leader
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try { 
                        	// available阻塞delay的时间，然后会被唤醒，这里就相当于一个延迟的效果了
                            available.awaitNanos(delay);
                        } finally { 
                        	// 将leader置为null，方便下一个元素出队
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
                // 当上面的逻辑都走完了之后，线程出队也被延迟了指定的delay时间，此时进入下一次循环，判断delay <= 0则可以出队了
            }
        } finally { 
        	// 当leader为null且队列中还有元素时，唤醒其他被阻塞的线程继续出队
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

通过出队操作的源码分析，我们就可以回答出leader和available的作用了

正在出队的线程就是leader线程，当leader线程调用了available.awaitNanos(delay)这句代码时，线程会释放锁，同时被挂起，然后其他的线程就可以尝试获取锁，然后进入出队的逻辑中，但是此时leader不为空，所以只能等着了，所以leader的作用就是标记当前正在出队的线程

available的话，则有两个作用，一个是防止队列为空时继续出队，另一个作用就是按照队列的顺序，阻塞住后面在排队出队的线程，得一个一个出队

来总结一下DelayQueue的出入队机制，首先入队时是加入到一个优先队列中，排序顺序就是按照我们定义的compare方法，一般按照延迟时间来排序，否则可能会导致最后出队的顺序不对；出队的话，是按照优先队列的顺序一个个出队，只有一个元素按照他的延迟时间出队了之后，下一个元素才能继续出队

DelayQueue保证线程安全使用的方式是加ReentrantLock，同时出队和入队都是使用的同一把锁，所以出入队操作不能同时执行

[PriorityBlockingQueue]: https://blog.csdn.net/LO_YUN/article/details/106057352