DelayQueue1.8源码

喜欢ヅ旅行 2022-04-25 02:38 176阅读 0赞

DelayQueue 内部通过组合PriorityQueue 来实现存储和维护元素顺序的，通过compareTo方法，排序存储，小的在前面。每次插入元素，都可能移动元素

DelayQueue 存储元素必须实现Delayed 接口，通过实现Delayed 接口，可以获取到元素延迟时间，以及可以比较元素大小（Delayed 继承Comparable）  
DelayQueue 通过一个可重入锁来控制元素的入队出队行为  
DelayQueue 中leader 标识 用于减少线程的竞争，表示当前有其它线程正在获取队头元素。  
PriorityQueue 只是负责存储数据以及维护元素的顺序，对于延迟时间取数据则是在DelayQueue 中进行判断控制的。  
DelayQueue 没有实现序列化接口

## 和时间轮算法区别 ##

之所以要用到PriorityQueue，主要是需要排序。也许后插入的消息需要比队列中的其他消息提前触发，那么这个后插入的消息就需要最先被消费者获取，这就需要排序功能。PriorityQueue内部使用最小堆来实现排序队列。队首的，最先被消费者拿到的就是最小的那个。使用最小堆让队列在数据量较大的时候比较有优势。使用最小堆来实现优先级队列主要是因为最小堆在插入和获取时，时间复杂度相对都比较好，都是O(logN)

时间轮是一种非常惊艳的数据结构。其在Linux内核中使用广泛，是Linux内核定时器的实现方法和基础之一。按使用场景，大致可以分为两种时间轮：原始时间轮和分层时间轮。分层时间轮是原始时间轮的升级版本，来应对时间“槽”数量比较大的情况，对内存和精度都有很高要求的情况。我们延迟任务的场景一般只需要用到原始时间轮就可以了。  
原始时间轮：如下图一个轮子，有8个“槽”，可以代表未来的一个时间。如果以秒为单位，中间的指针每隔一秒钟转动到新的“槽”上面，就好像手表一样。如果当前指针指在1上面，我有一个任务需要4秒以后执行，那么这个执行的线程回调或者消息将会被放在5上。那如果需要在20秒之后执行怎么办，由于这个环形结构槽数只到8，如果要20秒，指针需要多转2圈。位置是在2圈之后的5上面（20 % 8 + 1）。这个圈数需要记录在槽中的数据结构里面。这个数据结构最重要的是两个指针，一个是触发任务的函数指针，另外一个是触发的总第几圈数。时间轮可以用简单的数组或者是环形链表来实现。

![在这里插入图片描述][20190501151348315.png]

相比DelayQueue的数据结构，时间轮在算法复杂度上有一定优势。DelayQueue由于涉及到排序，需要调堆，插入和移除的复杂度是O(lgn)，而时间轮在插入和移除的复杂度都是O(1)。

## 使用 ##

消息消费时间不一定是制定的时间，例如下面

@Data
    public class Message implements Delayed {
        private int id;
        private String body; // 消息内容
        private long excuteTime;// 延迟时长，这个是必须的属性因为要按照这个判断延时时长。
    
        public Message(int id, String body, long delayTime) {
            this.id = id;
            this.body = body;
            this.excuteTime = TimeUnit.NANOSECONDS.convert(delayTime, TimeUnit.MILLISECONDS) + System.nanoTime();
        }
    
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(excuteTime - System.nanoTime(), TimeUnit.NANOSECONDS);
        }
    
        @Override
        public int compareTo(Delayed o) {
            Message msg = (Message) o;
            return this.id > msg.id ? 1
                    : this.id < msg.id ? -1 : 0;
        }
    
        public static void main(String[] args) throws InterruptedException {
            // 创建延时队列
            DelayQueue<Message> queue = new DelayQueue<>();
    
            // 添加延时消息,m1 延时3s
            Message m1 = new Message(3, "world", 2000);
            // 添加延时消息,m2 延时10s
            Message m2 = new Message(2, "hello", 4000);
            //将延时消息放到延时队列中
            queue.offer(m2);
            queue.offer(m1);
            System.out.println("消息入队 " + new Date());
    
            System.out.println("消费消息：" + queue.take() + " " + new Date());
            System.out.println("消费消息：" + queue.take() + " " + new Date());
    
        }
    
    }

消息入队 Tue Apr 30 16:07:52 CST 2019  
消费消息：Message(id=2, body=hello, excuteTime=5580123695174) Tue Apr 30 16:07:56 CST 2019  
消费消息：Message(id=3, body=world, excuteTime=5578123690265) Tue Apr 30 16:07:56 CST 2019

消息实现Delay接口

public interface Delayed extends Comparable<Delayed> {
    
        /**
         * Returns the remaining delay associated with this object, in the
         * given time unit.
         *
         * @param unit the time unit
         * @return the remaining delay; zero or negative values indicate
         * that the delay has already elapsed
         */
        long getDelay(TimeUnit unit);
    }
    
    public interface Comparable<T> {
        public int compareTo(T o);
    }

## DelayQueue构造器 ##

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
        implements BlockingQueue<E> {
    
        private final transient ReentrantLock lock = new ReentrantLock();
        private final PriorityQueue<E> q = new PriorityQueue<E>();
    
        /**
         * Thread designated to wait for the element at the head of
         * the queue.  This variant of the Leader-Follower pattern
         * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
         * minimize unnecessary timed waiting.  When a thread becomes
         * the leader, it waits only for the next delay to elapse, but
         * other threads await indefinitely.  The leader thread must
         * signal some other thread before returning from take() or
         * poll(...), unless some other thread becomes leader in the
         * interim.  Whenever the head of the queue is replaced with
         * an element with an earlier expiration time, the leader
         * field is invalidated by being reset to null, and some
         * waiting thread, but not necessarily the current leader, is
         * signalled.  So waiting threads must be prepared to acquire
         * and lose leadership while waiting.
         */
        private Thread leader = null;
    
        /**
         * Condition signalled when a newer element becomes available
         * at the head of the queue or a new thread may need to
         * become leader.
         */
        private final Condition available = lock.newCondition();
    
        /**
         * Creates a new {@code DelayQueue} that is initially empty.
         */
        public DelayQueue() {}

使用PriorityQueue存储，默认容量11

private static final int DEFAULT_INITIAL_CAPACITY = 11;
       public PriorityQueue() {
            this(DEFAULT_INITIAL_CAPACITY, null);
        }

## 存储元素 ##

/**
         * Inserts the specified element into this delay queue.
         *
         * @param e the element to add
         * @return {@code true}
         * @throws NullPointerException if the specified element is null
         */
        public boolean offer(E e) {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                q.offer(e);
                if (q.peek() == e) {
                    leader = null;
                    available.signal();
                }
                return true;
            } finally {
                lock.unlock();
            }
        }

PriorityQueue添加元素，不能为null

/**
         * Inserts the specified element into this priority queue.
         *
         * @return {@code true} (as specified by {@link Queue#offer})
         * @throws ClassCastException if the specified element cannot be
         *         compared with elements currently in this priority queue
         *         according to the priority queue's ordering
         * @throws NullPointerException if the specified element is null
         */
        public boolean offer(E e) {
            if (e == null)
                throw new NullPointerException();
            modCount++;
            int i = size;
            if (i >= queue.length)
                grow(i + 1);
            size = i + 1;
            if (i == 0)
                queue[0] = e;
            else
                siftUp(i, e);
            return true;
        }

第一个元素，queue\[0\] = e;结束  
第二个元素，siftUp(i, e);

/**
         * Inserts item x at position k, maintaining heap invariant by
         * promoting x up the tree until it is greater than or equal to
         * its parent, or is the root.
         *
         * To simplify and speed up coercions and comparisons. the
         * Comparable and Comparator versions are separated into different
         * methods that are otherwise identical. (Similarly for siftDown.)
         *
         * @param k the position to fill
         * @param x the item to insert
         */
        private void siftUp(int k, E x) {
            if (comparator != null)
                siftUpUsingComparator(k, x);
            else
                siftUpComparable(k, x);
        }

因为消息实现Comparable接口，comparator =null

@SuppressWarnings("unchecked")
        private void siftUpComparable(int k, E x) {
            Comparable<? super E> key = (Comparable<? super E>) x;
            while (k > 0) {
                int parent = (k - 1) >>> 1;//二叉树，孩子索引求父亲索引
                Object e = queue[parent];
                if (key.compareTo((E) e) >= 0)
                    break;
                queue[k] = e;
                k = parent;
            }
            queue[k] = key;
        }

因为第二个元素排序大，所以break;退出循环，直接后面添加元素  
如果第二个元素排序小，就会和第一个互换位置

PriorityQueue获取头元素，如果没有，返回null

@SuppressWarnings("unchecked")
        public E peek() {
            return (size == 0) ? null : (E) queue[0];
        }

## 取出元素 ##

/**
         * Retrieves and removes the head of this queue, waiting if necessary
         * until an element with an expired delay is available on this queue.
         *
         * @return the head of this queue
         * @throws InterruptedException {@inheritDoc}
         */
        public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                for (;;) {
                    E first = q.peek();
                    if (first == null)
                        available.await();//如果头结点为null，挂起线程
                    else {
                        long delay = first.getDelay(NANOSECONDS);
                        if (delay <= 0)
                            return q.poll();
                        first = null; // don't retain ref while waiting
                        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 && q.peek() != null)
                    available.signal();
                lock.unlock();
            }
        }

PriorityQueue-poll获取并删除第一个元素

@SuppressWarnings("unchecked")
        public E poll() {
            if (size == 0)
                return null;
            int s = --size;
            modCount++;
            E result = (E) queue[0];
            E x = (E) queue[s];
            queue[s] = null;
            if (s != 0)
                siftDown(0, x);
            return result;
        }

如果原来有两个以上元素，需要移动元素，把第二个元素移到第一个

/**
         * Inserts item x at position k, maintaining heap invariant by
         * demoting x down the tree repeatedly until it is less than or
         * equal to its children or is a leaf.
         *
         * @param k the position to fill
         * @param x the item to insert
         */
        private void siftDown(int k, E x) {
            if (comparator != null)
                siftDownUsingComparator(k, x);
            else
                siftDownComparable(k, x);
        }
    
    
        @SuppressWarnings("unchecked")
        private void siftDownComparable(int k, E x) {
            Comparable<? super E> key = (Comparable<? super E>)x;
            int half = size >>> 1;        // loop while a non-leaf
            while (k < half) {
                int child = (k << 1) + 1; // assume left child is least
                Object c = queue[child];
                int right = child + 1;
                if (right < size &&
                    ((Comparable<? super E>) c).compareTo((E) queue[right]) > 0)
                    c = queue[child = right];
                if (key.compareTo((E) c) <= 0)
                    break;
                queue[k] = c;
                k = child;
            }
            queue[k] = key;
        }

[20190501151348315.png]: /images/20220210/7be6b571070449c084dbeca5e57a90b1.png