Java 多线程（十）：DelayQueue、PriorityBlockingQueue、SynchronousQueue

Love The Way You Lie 2021-10-01 08:46 221阅读 0赞

## PriorityBlockingQueue ##

*  **PriorityBlockingQueue 是一个支持优先级的无边界阻塞队列**，默认情况下采用自然顺序排列，也可以通过比较器 Comparator 指定排序规则

public class Person {
        private String name;
        private int age;
    
        public Person(String name, int age) {
            this.name = name;
            this.age = age;
        }
    
        public String getName() {
            return name;
        }
    
        public int getAge() {
            return age;
        }
    
        @Override
        public String toString() {
            return "Person{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    '}';
        }
    }

public class Main {
        public static void main(String[] args) throws Exception{
            PriorityBlockingQueue<Person> queue = new PriorityBlockingQueue<Person>(3,((o1, o2) -> {
                if(o1.getAge() > o2.getAge())
                    return 1;
                else if(o1.getAge()<o2.getAge())
                    return -1;
                return 0;
            }));
            queue.put(new Person("a",3));
            queue.put(new Person("b",1));
            queue.put(new Person("c",2));
            System.out.println(queue.take());
            System.out.println(queue.take());
            System.out.println(queue.take());
        }
    }

## DelayQueue ##

*  **DelayQueue 是一个支持延迟获取元素的无边界阻塞队列**，列头的元素是最先到期的元素，若无元素到期，即使队列有元素，也无法从列头获取元素
 *  放入队列的元素需要实现 Delayed 接口，Delayed 接口继承了 Comparable 接口，所以需要实现下面两个方法：
    
     *  **long getDelay(TimeUnit unit)**：获取元素剩余的延迟时间，当返回值小于 0 时，则视为到期
     *  **int compareTo(Delayed o)**：用于延迟队列内部比较排序

public class DelayedElement implements Delayed {
    
        private long delay; //延迟时间
        private long expire;  //到期时间
        private int number;   //值
        private long now; //创建时间
    
        public DelayedElement(long delay, int number) {
            this.delay = delay;
            this.number = number;
            expire = System.currentTimeMillis() + delay;    //到期时间 = 当前时间+延迟时间
            now = System.currentTimeMillis();
        }
    
        /**
         * 需要实现的接口，获得延迟时间   用过期时间-当前时间
         */
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    
        /**
         * 用于延迟队列内部比较排序   
         */
        @Override
        public int compareTo(Delayed o) {
            DelayedElement delayedElement = (DelayedElement) o;
            return this.number - ((DelayedElement) o).getNumber();
        }
    
        public int getNumber() {
            return number;
        }
    
        @Override
        public String toString() {
            return "DelayedElement{" +
                    "delay=" + delay +
                    ", expire=" + expire +
                    ", number=" + number +
                    ", now=" + now +
                    '}';
        }
    }

public class Main2 {
        public static void main(String[] args) throws Exception{
            DelayQueue delayQueue = new DelayQueue();
    
           DelayedElement d1 = new DelayedElement(1000,1);
           DelayedElement d2 = new DelayedElement(1000,3);
           DelayedElement d3 = new DelayedElement(1000,2);
    
           delayQueue.put(d1);
           delayQueue.put(d2);
           delayQueue.put(d3);
    
           System.out.println(delayQueue.take());
            System.out.println(delayQueue.take());
            System.out.println(delayQueue.take());
        }
    }

#### SynchronousQueue ####

*  SynchronousQueue 实际上不是一个真正的队列，因为它并不能存储元素，因此 put 和 take 会一直阻塞当前线程，每一个插入操作都必须等待另一个线程的删除操作

public class Main2 {
        public static void main(String[] args) {
            SynchronousQueue<String> syncQueue = new SynchronousQueue<>();
    
            new Thread(()->{
                try {
                    syncQueue.put("a");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
    
            new Thread(()->{
                try {
                    System.out.println(syncQueue.take());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }