并发队列之DelayQueue

  已经说了四个并发队列了,DelayQueue这是最后一个,这是一个无界阻塞延迟队列,底层基于前面说过的PriorityBlockingQueue实现的 ,队列中每个元素都有过期时间,当从队列获取元素时,只有过...

  已经说了四个并发队列了,DelayQueue这是最后一个,这是一个无界阻塞延迟队列,底层基于前面说过的PriorityBlockingQueue实现的 ,队列中每个元素都有过期时间,当从队列获取元素时,只有过期元素才会出队列,而队列头部的元素是过期最快的元素;

一.简单使用

  可以看到我们可以自己设置超时时间和优先级队列中的比较规则,这样我们在队列中取的时候,按照最快超时的先出队;

package com.example.demo.study;

import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

import lombok.Data;

public class Study0210 {

    @Data
    static class MyDelayed implements Delayed {
        private long delayTime;//该任务需要再队列中的延迟的时候
        private long expire;//这个时间表示当前时间和延迟时间相加,这里就叫做到期时间
        private String taskName;//任务的名称

        public MyDelayed(long delayTime, String taskName) {
            this.delayTime = delayTime;
            this.taskName = taskName;
            this.expire = System.currentTimeMillis()+delayTime;
        }
        
        //指定优先级队列里面的比较规则,就跟上篇博客中说的优先级队列中说的比较器一样
        @Override
        public int compareTo(Delayed o) {
            return (int)(this.getDelay(TimeUnit.MILLISECONDS)-o.getDelay(TimeUnit.MILLISECONDS));
        }

        //这个方法表示该任务在队列中还有多少剩余时间,也就是expire-当前时间
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.expire-System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        //创建延迟队列
        DelayQueue<MyDelayed> queue = new DelayQueue<MyDelayed>();
        
        //创建任务丢到队列中
        Random random = new Random();
        for (int i = 1; i < 11; i++) {
            MyDelayed myDelayed = new MyDelayed(random.nextInt(500),"task"+i);
            queue.add(myDelayed);
        }
        
        //获取队列中的任务,这里只会跟超时时间最小的有关,和入队顺序无关
        MyDelayed myDelayed = queue.take();
        while(myDelayed!=null) {
            System.out.println(myDelayed.toString());
            myDelayed = queue.take();
        }
    }
}

二.基本组成 

//由此可是这个队列中存放的任务必须是Delayed类型的
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>();
    //leader线程,实际上每次进行入队和出队操作的只能是leader线程,其余的都叫做fallower线程,这里会用到一个leader-follower模式
    private Thread leader = null;
    //条件变量
    private final Condition available = lock.newCondition();

    //省略很多代码
}

  

  具体的继承关系可以看看下面这个图,实际操作的都是内部的PriorityQueue;

三.offer方法

  上面代码中我们虽然说调用的是add方法,其实就是调用的是offer方法;

public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    //获取锁
    lock.lock();
    try {
        //往优先级队列中添加一个元素
        q.offer(e);
        //注意,peek方法只是获取优先级队列中第一个元素,并不会删除
        //如果优先级队列中取的元素就是和当前添加的元素一样,说明当前元素就是达到过期要求的,于是设置leader线程为null
        //然后通知条件队列中的线程优先级队列中已经有元素了,可以过来取了
        if (q.peek() == e) {
            leader = null;
            available.signal();
        }
        return true;
    } finally {
        //释放锁
        lock.unlock();
    }
}

四.take方法

  获取并移除队列中达到超时时间要求的元素,如果队列中没有元素,就把当前线程丢到条件队列中阻塞;

  从下面的代码逻辑中我们可以知道:线程分为两种,一种是leader线程,一种是follower线程,其中leader线程只会阻塞一定的时间,follower线程会在条件队列阻塞无限长的时间;当leader线程执行完take操作之后,就会重置leader线程为null,然后从条件队列中拿一个出来设置为leader线程

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    //获取锁,可中断
    lock.lockInterruptibly();
    try {
        for (;;) {
            //这里先是尝试从优先级队列中获取一下节点,获取不到的话,说明当前优先级队列为空,就阻塞当前线程
            E first = q.peek();
            if (first == null)
                available.await();
            else {
                //如果优先级队列中有元素,那么肯定能走到这里来,然后取到该元素的超时时间,如果小于0,说明已经达到要求了,可以获取并删除队列中的元素
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    return q.poll();
                first = null; // don't retain ref while waiting
                //如果leader队列不为空,说明有其他线程正在执行take,于是就把当前线程放到条件队列中
                if (leader != null)
                    available.await();
                //到这里,说明优先级队列中没有元素到超时时间,而且此时没有其他线程调用take方法,于是就把leader线程设置为当前线程,
                //然后当前leader线程就会等待一定的时间,等优先级队列中最快超时的元素;
                //在等待的时候,leader线程会释放锁,这时其他线程B可以调用offer方法添加元素,线程C也可以调用take方法,然后线程C就会在
                //上面这里阻塞无限长的时间,直到被唤醒
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        available.awaitNanos(delay);
                    } finally {
                        //当前线程阻塞一定时间之后,不管成功了没有,都会把leader线程重置为null,然后重新循环
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    //这里的意思就是当前线程移除元素成功之后,唤醒条件队列中的线程去继续从队列中获取元素
    } finally {
        if (leader == null && q.peek() != null)
            available.signal();
        //释放锁
        lock.unlock();
    }
}

五.poll操作

  获取并移除队头过期元素,如果队列为空,或者对头元素没有过超时时间就返回null

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        //尝试获取队头元素,如果队头元素为空或者该延迟过期时间还没到,就返回null
        E first = q.peek();
        if (first == null || first.getDelay(NANOSECONDS) > 0)
            return null;
        else
        //否则就获取并移除队头元素
            return q.poll();
    } finally {
        lock.unlock();
    }
}

六.总结

  这个队列其实很容易,主要的是有一个延迟时间,我们从优先级队列中获取的根节点首先会判断有没有过超时时间,有的话就移除并返回就好了,没有的话,就看看还剩下多少时间才会超时(由于是优先级队列,所以根节点一般就是最快超时时间的,当然,也可以修改优先级队列的比较规则),于是当前线程就会等这个节点超时,此时leader等于当前线程,在等待的过程中,会释放锁,所以其他线程可以往队列中添加元素,也可以获取元素(但是由于此时leader!=null,这些线程会阻塞无限长时间直到被唤醒);

  在leader线程超时时间到了之后自动唤醒,再进行一次循环,就会获取并移除根节点了,最后再重置leader节点为null,顺便唤醒条件队列中的节点;

  • 发表于 2020-02-11 22:01
  • 阅读 ( 113 )
  • 分类:网络文章

条评论

请先 登录 后评论
不写代码的码农
小编

篇文章

作家榜 »

  1. 小编 文章
返回顶部
部分文章转自于网络,若有侵权请联系我们删除